[Simulator] Add high-fidelity CPU-based inference simulator (#33824)

Co-authored-by: zhouhaizhu.zhz <zhouhaizhu.zhz@alibaba-inc.com>
Co-authored-by: LinSiyuan814 <linsiyuan.lsy@alibaba-inc.com>
Co-authored-by: hzh0425 <hzh0425@apache.org>

tools/sglang-simulator/test/assets/qwen3-8b/config.json

@@ -0,0 +1,28 @@
+{
+  "architectures": ["Qwen3ForCausalLM"],
+  "attention_bias": false,
+  "attention_dropout": 0.0,
+  "bos_token_id": 151643,
+  "eos_token_id": 151645,
+  "head_dim": 128,
+  "hidden_act": "silu",
+  "hidden_size": 4096,
+  "initializer_range": 0.02,
+  "intermediate_size": 12288,
+  "max_position_embeddings": 40960,
+  "max_window_layers": 36,
+  "model_type": "qwen3",
+  "num_attention_heads": 32,
+  "num_hidden_layers": 36,
+  "num_key_value_heads": 8,
+  "rms_norm_eps": 1e-06,
+  "rope_scaling": null,
+  "rope_theta": 1000000,
+  "sliding_window": null,
+  "tie_word_embeddings": false,
+  "torch_dtype": "bfloat16",
+  "transformers_version": "5.12.1",
+  "use_cache": true,
+  "use_sliding_window": false,
+  "vocab_size": 151936
+}

tools/sglang-simulator/test/test_simulation_cache_hit_ratio.py

@@ -0,0 +1,68 @@
+import json
+
+import pytest
+from sglang_simulator.simulation.benchmark import BenchmarkConfig
+from test_simulation_sglang_runner import make_fixed_dataset, make_sglang_runner
+from test_simulation_sglang_serving import (
+    SIM_CONFIGS,
+    SGLangServingRunner,
+    assert_decode_metrics,
+)
+
+
+def test_in_process_runner_reports_each_cache_tier(tmp_path):
+    runner = make_sglang_runner(tmp_path)
+    benchmark_config = BenchmarkConfig(request_rate=10, ignore_request_timestamp=False)
+    cached_ds = make_fixed_dataset(1000, 8)
+    evict_l1_ds = make_fixed_dataset(2000, 10)
+    evict_l2_ds = make_fixed_dataset(3000, 20)
+
+    try:
+        metrics = runner.benchmark(benchmark_config, dataset=cached_ds)
+        assert metrics["completed"] == len(cached_ds)
+        assert metrics["prefix_cache_reused_ratio"] == 0
+
+        metrics = runner.benchmark(benchmark_config, dataset=cached_ds)
+        assert metrics["kv_cache_device_hit_ratio"] > 0.95
+
+        runner.benchmark(benchmark_config, dataset=evict_l1_ds)
+        metrics = runner.benchmark(benchmark_config, dataset=cached_ds)
+        assert metrics["kv_cache_host_hit_ratio"] > 0.95
+
+        runner.benchmark(benchmark_config, dataset=evict_l2_ds)
+        metrics = runner.benchmark(benchmark_config, dataset=cached_ds)
+        assert metrics["kv_cache_storage_hit_ratio"] > 0.95
+    finally:
+        runner.shutdown()
+
+
+def test_second_replay_benchmark_hits_all_reusable_prefix_tokens(tmp_path, monkeypatch):
+    # This test validates cache reuse across consecutive benchmark runs.
+    monkeypatch.setenv("SGLANG_IS_IN_CI", "false")
+
+    runner = SGLangServingRunner(SIM_CONFIGS["replay"], tmp_path)
+    try:
+        first_metrics = runner.benchmark(tmp_path / "benchmark-first.json")
+        second_metrics = runner.benchmark(tmp_path / "benchmark-second.json")
+    finally:
+        runner.shutdown()
+
+    assert_decode_metrics(first_metrics)
+    assert_decode_metrics(second_metrics)
+
+    assert second_metrics["total_input"] == 24
+    assert second_metrics["total_new_input"] == 3
+    assert second_metrics["prefix_cache_reused_ratio"] == pytest.approx(0.875)
+    assert second_metrics["kv_cache_device_hit_ratio"] == pytest.approx(0.875)
+    assert second_metrics["kv_cache_host_hit_ratio"] == 0
+    assert second_metrics["kv_cache_storage_hit_ratio"] == 0
+
+    requests = [
+        json.loads(line)
+        for line in (runner.output_dir / "request.jsonl")
+        .read_text(encoding="utf-8")
+        .splitlines()
+    ]
+    assert len(requests) == 3
+    assert all(request["input_length"] == 8 for request in requests)
+    assert all(request["final_device_hit_len"] == 7 for request in requests)

tools/sglang-simulator/test/test_simulation_offline_blocking.py

@@ -0,0 +1,80 @@
+import json
+
+import pytest
+from test_simulation_sglang_serving import (
+    SIM_CONFIGS,
+    SGLangServingRunner,
+    assert_decode_metrics,
+)
+
+REQUEST_RATE = 1
+SEED = 123
+RELATIVE_TOLERANCES = {
+    "duration": 0.01,
+    "request_throughput": 0.01,
+    "input_throughput": 0.01,
+    "output_throughput": 0.01,
+    "mean_e2e_latency_ms": 0.10,
+    "mean_ttft_ms": 0.10,
+    "mean_tpot_ms": 0.10,
+    "mean_itl_ms": 0.10,
+}
+
+
+def _relative_error(actual, expected):
+    return abs(actual - expected) / abs(expected)
+
+
+def _run_mode(mode, tmp_path):
+    case_dir = tmp_path / mode
+    case_dir.mkdir()
+    runner = SGLangServingRunner(SIM_CONFIGS["aic_sol"], case_dir, mode=mode)
+    try:
+        metrics = runner.benchmark(
+            case_dir / "benchmark.json", request_rate=REQUEST_RATE, seed=SEED
+        )
+    finally:
+        runner.shutdown()
+
+    requests = [
+        json.loads(line)
+        for line in (runner.output_dir / "request.jsonl")
+        .read_text(encoding="utf-8")
+        .splitlines()
+    ]
+    requests.sort(key=lambda request: request["created_time"])
+    return metrics, requests
+
+
+def test_request_rate_offline_matches_blocking(tmp_path):
+    offline_metrics, offline_requests = _run_mode("offline", tmp_path)
+    blocking_metrics, blocking_requests = _run_mode("blocking", tmp_path)
+
+    for metrics in (offline_metrics, blocking_metrics):
+        assert_decode_metrics(metrics)
+
+    assert len(offline_requests) == len(blocking_requests) == 3
+
+    offline_arrivals = [request["created_time"] for request in offline_requests]
+    blocking_arrivals = [request["created_time"] for request in blocking_requests]
+    assert offline_arrivals[1] > 0.5
+    assert blocking_arrivals[1] > 0.5
+    assert offline_arrivals == pytest.approx(blocking_arrivals, abs=0.02)
+    assert (
+        offline_metrics["max_concurrent_requests"]
+        == blocking_metrics["max_concurrent_requests"]
+        == 1
+    )
+
+    for key in ("completed", "total_input", "total_output"):
+        assert offline_metrics[key] == blocking_metrics[key]
+
+    for key, tolerance in RELATIVE_TOLERANCES.items():
+        error = _relative_error(offline_metrics[key], blocking_metrics[key])
+        assert error <= tolerance, (
+            key,
+            offline_metrics[key],
+            blocking_metrics[key],
+            error,
+            tolerance,
+        )

tools/sglang-simulator/test/test_simulation_sglang_runner.py

@@ -0,0 +1,118 @@
+import atexit
+import json
+import os
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+from sglang_simulator.dataset import GenericRequest, SimpleDataset
+from sglang_simulator.simulation.benchmark import BenchmarkConfig
+
+ASSETS = Path(__file__).parent / "assets"
+SGLANG_ROOT = Path(__file__).parents[3]
+if str(SGLANG_ROOT) not in sys.path:
+    sys.path.insert(0, str(SGLANG_ROOT))
+os.environ.setdefault("CUDA_VISIBLE_DEVICES", "")
+
+
+def make_fixed_dataset(
+    start_token: int,
+    count: int,
+    *,
+    input_length: int = 1025,
+    output_length: int = 1,
+) -> SimpleDataset:
+    return SimpleDataset(
+        reqs=[
+            GenericRequest(
+                token_ids=[start_token + i] * input_length,
+                input_length=input_length,
+                output_length=output_length,
+                custom_params={"created_time": i / 10},
+            )
+            for i in range(count)
+        ]
+    )
+
+
+def _write_sim_config(tmp_path: Path) -> Path:
+    table_path = tmp_path / "replay.json"
+    table_path.write_text(
+        json.dumps({"[[1, 1024]]": 0.001, "[[1025, 0]]": 0.01}),
+        encoding="utf-8",
+    )
+    config = {
+        "platform": {
+            "accelerator": {"name": "a100_sxm", "hbm_capacity_gb": 80},
+            "disk_read_bandwidth_gb": 8,
+            "disk_write_bandwidth_gb": 8,
+            "memory_read_bandwidth_gb": 64,
+            "memory_write_bandwidth_gb": 64,
+            "num_device_per_node": 8,
+        },
+        "predictor": {
+            "name": "replay",
+            "database_path": str(table_path),
+            "miss_strategy": "knn",
+            "miss_knn_k": 1,
+        },
+        "scheduler": {"tp_size": 1, "ep_size": 1, "dp_size": 1},
+    }
+    config_path = tmp_path / "sim_config.json"
+    config_path.write_text(json.dumps(config), encoding="utf-8")
+    return config_path
+
+
+def make_sglang_runner(tmp_path: Path):
+    os.environ["SGLANG_SIMULATOR_CONFIG_PATH"] = str(_write_sim_config(tmp_path))
+
+    from benchmark.simulator.bench_runner import SGLangBenchmarkRunner
+    from sglang.srt.server_args import ServerArgs
+
+    runner = SGLangBenchmarkRunner(
+        server_args=ServerArgs(
+            model_path=str(ASSETS / "qwen3-8b"),
+            load_format="dummy",
+            device="cpu",
+            enable_hierarchical_cache=True,
+            hicache_ratio=2,
+            hicache_storage_backend="file",
+            hicache_storage_prefetch_policy="wait_complete",
+            max_total_tokens=10 * 1024,
+            page_size=256,
+            skip_tokenizer_init=True,
+        )
+    )
+    runner.clear_hicache_storage()
+    return runner
+
+
+def test_benchmark_sglang_runs_paged_decode(tmp_path):
+    runner = make_sglang_runner(tmp_path)
+    dataset = make_fixed_dataset(
+        1000,
+        2,
+        input_length=1024,
+        output_length=2,
+    )
+
+    try:
+        metrics = runner.benchmark(
+            BenchmarkConfig(request_rate=10, ignore_request_timestamp=False),
+            dataset=dataset,
+        )
+        request_stats = runner.get_request_stats()
+    finally:
+        with patch.object(atexit, "unregister", wraps=atexit.unregister) as unregister:
+            runner.shutdown()
+            runner.shutdown()
+
+        unregister.assert_called_once_with(runner.engine.shutdown)
+
+    assert metrics["completed"] == len(dataset)
+    assert metrics["total_input"] == 2 * 1024
+    assert metrics["total_output"] == 2 * 2
+    assert metrics["mean_tpot_ms"] > 0
+    assert all(
+        idx == 0 or req["created_time"] > 0 for idx, req in enumerate(request_stats)
+    )

tools/sglang-simulator/test/test_simulation_sglang_serving.py

@@ -0,0 +1,163 @@
+import json
+import os
+import signal
+import socket
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+import requests
+
+ASSETS = Path(__file__).parent / "assets"
+SGLANG_ROOT = Path(__file__).parents[3]
+BENCH_SERVING = SGLANG_ROOT / "benchmark" / "simulator" / "bench_serving.py"
+EXAMPLES = Path(__file__).parent.parent / "examples"
+SIM_CONFIGS = {
+    "aic_sol": EXAMPLES / "sim_configs" / "aic_sol.json",
+    "aic_silicon": EXAMPLES / "sim_configs" / "aic_silicon.json",
+    "ml": EXAMPLES / "sim_configs" / "ml.json",
+    "replay": EXAMPLES / "sim_configs" / "replay.json",
+}
+
+
+class SGLangServingRunner:
+    def __init__(self, config_path: Path, tmp_path: Path, mode: str = "offline"):
+        self.mode = mode
+        with socket.socket() as sock:
+            sock.bind(("127.0.0.1", 0))
+            self.port = sock.getsockname()[1]
+
+        self.output_dir = tmp_path / "output"
+        env = os.environ.copy()
+        env.update(
+            CUDA_VISIBLE_DEVICES="",
+            SGLANG_USE_CPU_ENGINE="1",
+            SGLANG_SIMULATOR_CONFIG_PATH=str(config_path),
+            SGLANG_SIMULATOR_OUTPUT_MODE=mode.upper(),
+            SGLANG_SIMULATOR_OUTPUT_DIR=str(self.output_dir),
+        )
+        cmd = [
+            sys.executable,
+            "-m",
+            "sglang_simulator.simulation.sglang.launch_server",
+            "--model-path",
+            str(ASSETS / "qwen3-8b"),
+            "--sim-config-path",
+            str(config_path),
+            "--port",
+            str(self.port),
+            "--tokenizer-path",
+            str(EXAMPLES / "assets" / "tokenizer"),
+            "--max-total-tokens",
+            "8192",
+            "--max-running-requests",
+            "8",
+            "--disable-overlap-schedule",
+        ]
+        self.server_proc = subprocess.Popen(cmd, env=env, preexec_fn=os.setsid)
+        for _ in range(120):
+            if self.server_proc.poll() is not None:
+                raise RuntimeError("SGLang Simulator server exited during startup")
+            try:
+                if requests.get(self.base_url, timeout=1).status_code < 500:
+                    return
+            except requests.RequestException:
+                pass
+            time.sleep(1)
+        self.shutdown()
+        raise RuntimeError("SGLang Simulator server did not become ready")
+
+    @property
+    def base_url(self) -> str:
+        return f"http://127.0.0.1:{self.port}"
+
+    def benchmark(
+        self,
+        output_file: Path,
+        workload: str = "sharegpt",
+        request_rate=None,
+        seed=42,
+    ) -> dict:
+        cmd = [
+            sys.executable,
+            str(BENCH_SERVING),
+            f"--simulator-mode={self.mode}",
+            "--backend=sglang",
+            f"--base-url={self.base_url}",
+            f"--model={ASSETS / 'qwen3-8b'}",
+            f"--tokenizer={EXAMPLES / 'assets' / 'tokenizer'}",
+            "--num-prompts=3",
+            "--disable-tqdm",
+            "--profile",
+            f"--output-file={output_file}",
+        ]
+        if request_rate is not None:
+            cmd.extend([f"--request-rate={request_rate}", f"--seed={seed}"])
+
+        if workload == "sharegpt":
+            cmd.extend(
+                [
+                    "--dataset-name=sharegpt",
+                    f"--dataset-path={EXAMPLES / 'workloads' / 'sharegpt-example.json'}",
+                    "--sharegpt-output-len=4",
+                ]
+            )
+        else:
+            assert workload == "timestamp_trace"
+            cmd.extend(
+                [
+                    "--dataset-name=autobench",
+                    f"--dataset-path={EXAMPLES / 'workloads' / 'timestamp-trace-example.jsonl'}",
+                    "--use-trace-timestamps",
+                ]
+            )
+
+        subprocess.run(cmd, check=True)
+        assert output_file.is_file()
+        return json.loads(
+            (self.output_dir / "metrics.json").read_text(encoding="utf-8")
+        )
+
+    def shutdown(self):
+        if self.server_proc.poll() is not None:
+            return
+        os.killpg(self.server_proc.pid, signal.SIGTERM)
+        try:
+            self.server_proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            os.killpg(self.server_proc.pid, signal.SIGKILL)
+            self.server_proc.wait()
+
+
+def assert_decode_metrics(metrics):
+    assert metrics["completed"] == 3
+    assert metrics["total_output"] == 12
+    assert metrics["mean_ttft_ms"] >= 0
+    assert metrics["mean_tpot_ms"] > 0
+    assert metrics["mean_itl_ms"] > 0
+    assert metrics["input_throughput"] > 0
+
+
+@pytest.mark.parametrize("config_name", SIM_CONFIGS)
+def test_benchmark(config_name, tmp_path):
+    runner = SGLangServingRunner(SIM_CONFIGS[config_name], tmp_path)
+    try:
+        metrics = runner.benchmark(tmp_path / "benchmark.json")
+    finally:
+        runner.shutdown()
+
+    assert_decode_metrics(metrics)
+
+
+def test_timestamp_trace(tmp_path):
+    runner = SGLangServingRunner(SIM_CONFIGS["replay"], tmp_path)
+    try:
+        metrics = runner.benchmark(
+            tmp_path / "benchmark.json", workload="timestamp_trace"
+        )
+    finally:
+        runner.shutdown()
+
+    assert_decode_metrics(metrics)