Spaces:
Running
Running
File size: 1,424 Bytes
0c6c82c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 | from inferscale import run_simulation
def base_config():
return {
"duration_s": 8,
"request_rate_rps": 1.5,
"prompt_tokens_mean": 128,
"prompt_tokens_cv": 0.1,
"output_tokens_mean": 16,
"output_tokens_cv": 0.1,
"seed": 3,
}
def test_simulation_completes_requests():
result = run_simulation(base_config())
assert result["summary"]["requests_generated"] > 0
assert result["summary"]["requests_unfinished"] == 0
assert result["latency"]["ttft_ms"]["p95"] > 0
assert result["resource"]["peak_kv_gb"] >= 0
assert result["provenance"]["latency_profile_type"] == "analytical-reference"
def test_high_load_increases_tail_latency():
low = base_config()
high = base_config()
low["request_rate_rps"] = 0.5
high["request_rate_rps"] = 8.0
a = run_simulation(low)
b = run_simulation(high)
assert b["latency"]["ttft_ms"]["p95"] >= a["latency"]["ttft_ms"]["p95"]
def test_static_and_continuous_are_distinct():
cfg = base_config()
cfg.update({"request_rate_rps": 4.0, "output_tokens_mean": 32})
cfg["scheduler"] = "static_fcfs"
static = run_simulation(cfg)
cfg["scheduler"] = "continuous_fcfs"
continuous = run_simulation(cfg)
assert static["summary"]["goodput_rps"] != continuous["summary"]["goodput_rps"] or static["latency"]["ttft_ms"]["p95"] != continuous["latency"]["ttft_ms"]["p95"]
|