from __future__ import annotations from copy import deepcopy from statistics import mean, pstdev from .models import SimulationConfig from .simulator import run_simulation def evaluate_rate(base: SimulationConfig, rate: float, repetitions: int = 3) -> dict: runs = [] for rep in range(repetitions): cfg = deepcopy(base) cfg.request_rate_rps = rate cfg.seed = base.seed + rep * 101 runs.append(run_simulation(cfg.to_dict())) attainments = [run["summary"]["slo_attainment"] for run in runs] goodputs = [run["summary"]["goodput_rps"] for run in runs] ttfts = [run["latency"]["ttft_ms"]["p95"] for run in runs] e2es = [run["latency"]["e2e_ms"]["p95"] for run in runs] unfinished_values = [run["summary"]["requests_unfinished"] for run in runs] passed = all( attainment >= base.slo_attainment_target and unfinished == 0 for attainment, unfinished in zip(attainments, unfinished_values, strict=True) ) return { "rate_rps": rate, "passed": passed, "slo_attainment": mean(attainments), "slo_attainment_min": min(attainments), "slo_attainment_max": max(attainments), "slo_attainment_std": pstdev(attainments) if len(attainments) > 1 else 0.0, "goodput_rps": mean(goodputs), "goodput_std": pstdev(goodputs) if len(goodputs) > 1 else 0.0, "p95_ttft_ms": mean(ttfts), "p95_e2e_ms": mean(e2es), "mean_unfinished": mean(unfinished_values), "max_unfinished": max(unfinished_values), "repetitions": repetitions, "criterion": "all-repetitions-meet-target-and-drain", "target": base.slo_attainment_target, } def capacity_search( config: dict, min_rate: float = 0.25, max_rate: float = 32.0, iterations: int = 8, repetitions: int = 2, headroom: float = 0.20, ) -> dict: base = SimulationConfig.from_dict(config) low = max(0.01, min_rate) high = max(low * 1.01, max_rate) trace: list[dict] = [] low_eval = evaluate_rate(base, low, repetitions) trace.append(low_eval) if not low_eval["passed"]: return { "status": "no_feasible_rate", "capacity_rps": 0.0, "recommended_rps": 0.0, "headroom": headroom, "criterion": "all-repetitions-meet-target-and-drain", "trace": trace, } high_eval = evaluate_rate(base, high, repetitions) trace.append(high_eval) if high_eval["passed"]: return { "status": "upper_bound_still_feasible", "capacity_rps": high, "recommended_rps": high * (1.0 - headroom), "headroom": headroom, "criterion": "all-repetitions-meet-target-and-drain", "trace": sorted(trace, key=lambda item: item["rate_rps"]), } best = low for _ in range(max(iterations, 1)): mid = (low + high) / 2.0 result = evaluate_rate(base, mid, repetitions) trace.append(result) if result["passed"]: best = mid low = mid else: high = mid return { "status": "ok", "capacity_rps": best, "recommended_rps": best * (1.0 - headroom), "headroom": headroom, "criterion": "all-repetitions-meet-target-and-drain", "trace": sorted(trace, key=lambda item: item["rate_rps"]), } def compare_schedulers(config: dict, schedulers: list[str] | None = None) -> dict: schedulers = schedulers or [ "static_fcfs", "continuous_fcfs", "continuous_sjf", "continuous_slo", "chunked_slo", ] base = SimulationConfig.from_dict(config) base.topology = "colocated" rows = [] for scheduler in schedulers: cfg = deepcopy(base) cfg.scheduler = scheduler result = run_simulation(cfg.to_dict()) rows.append( { "scheduler": scheduler, "request_throughput_rps": result["summary"]["request_throughput_rps"], "goodput_rps": result["summary"]["goodput_rps"], "slo_attainment": result["summary"]["slo_attainment"], "p95_ttft_ms": result["latency"]["ttft_ms"]["p95"], "p95_e2e_ms": result["latency"]["e2e_ms"]["p95"], "peak_kv_utilization": result["resource"]["peak_kv_utilization"], "unfinished": result["summary"]["requests_unfinished"], "bottleneck": result["diagnostics"]["label"], } ) rows.sort(key=lambda row: (row["slo_attainment"], row["goodput_rps"]), reverse=True) return {"rows": rows} def compare_topologies(config: dict) -> dict: """Compare colocated and P/D-disaggregated serving on one deterministic trace.""" base = SimulationConfig.from_dict(config) rows = [] scenarios = [ ("colocated", False), ("colocated", True), ("disaggregated_pd", False), ("disaggregated_pd", True), ] for topology, cache_enabled in scenarios: cfg = deepcopy(base) cfg.topology = topology cfg.prefix_cache_enabled = cache_enabled if topology == "disaggregated_pd" and cfg.scheduler == "static_fcfs": cfg.scheduler = "continuous_fcfs" result = run_simulation(cfg.to_dict()) resource = result["resource"] accelerator_instances = int(resource.get("accelerator_instances", 1)) goodput_rps = result["summary"]["goodput_rps"] rows.append( { "scenario": f"{topology}{'_cache' if cache_enabled else ''}", "topology": topology, "prefix_cache": cache_enabled, "accelerator_instances": accelerator_instances, "goodput_rps": goodput_rps, "goodput_per_accelerator": goodput_rps / max(accelerator_instances, 1), "request_throughput_rps": result["summary"]["request_throughput_rps"], "slo_attainment": result["summary"]["slo_attainment"], "p95_ttft_ms": result["latency"]["ttft_ms"]["p95"], "p95_e2e_ms": result["latency"]["e2e_ms"]["p95"], "peak_kv_utilization": resource.get("peak_kv_utilization", 0.0), "prefix_hit_rate": resource.get("prefix_cache_hit_rate", 0.0), "prefill_tokens_saved": resource.get("prefill_tokens_saved", 0), "p95_transfer_ms": resource.get("p95_transfer_ms", 0.0), "prefill_busy_fraction": resource.get("prefill_busy_fraction", 0.0), "decode_busy_fraction": resource.get("decode_busy_fraction", result["summary"].get("busy_fraction", 0.0)), "transfer_busy_fraction": resource.get("transfer_busy_fraction", 0.0), "bottleneck": result["diagnostics"]["label"], } ) rows.sort(key=lambda row: (row["slo_attainment"], row["goodput_rps"]), reverse=True) return {"rows": rows} def _is_dominated(candidate: dict, rows: list[dict], throughput_key: str = "goodput_rps") -> bool: for other in rows: if other is candidate: continue no_worse = ( other[throughput_key] >= candidate[throughput_key] and other["p95_ttft_ms"] <= candidate["p95_ttft_ms"] ) strictly_better = ( other[throughput_key] > candidate[throughput_key] or other["p95_ttft_ms"] < candidate["p95_ttft_ms"] ) if no_worse and strictly_better: return True return False def design_space_search(config: dict, include_disaggregated: bool = True) -> dict: """Small browser-safe design-space sweep with a goodput/TTFT Pareto frontier. The sweep is intentionally bounded: its purpose is to expose configuration interactions interactively, not to claim exhaustive optimization. """ base = SimulationConfig.from_dict(config) rows: list[dict] = [] colocated_candidates = [] for scheduler in ["continuous_fcfs", "continuous_slo", "chunked_slo"]: for batch in [8, 16, 32]: colocated_candidates.append((scheduler, batch, False)) # Prefix reuse is added as a separate systems dimension for the SLO-aware # scheduler at the user's current batch size. colocated_candidates.append(("continuous_slo", base.max_batch_size, True)) for scheduler, batch, cache in colocated_candidates: cfg = deepcopy(base) cfg.topology = "colocated" cfg.scheduler = scheduler cfg.max_batch_size = batch cfg.prefix_cache_enabled = cache result = run_simulation(cfg.to_dict()) rows.append(_design_row(result, cfg, f"colocated / {scheduler} / batch {batch}{' / cache' if cache else ''}")) if include_disaggregated: for prefill_workers, decode_workers in [(1, 1), (1, 2), (2, 1)]: for cache in [False, True]: cfg = deepcopy(base) cfg.topology = "disaggregated_pd" cfg.scheduler = "continuous_slo" cfg.prefill_workers = prefill_workers cfg.decode_workers = decode_workers cfg.prefix_cache_enabled = cache result = run_simulation(cfg.to_dict()) label = f"P/D {prefill_workers}P:{decode_workers}D{' / cache' if cache else ''}" rows.append(_design_row(result, cfg, label)) for row in rows: row["pareto"] = not _is_dominated(row, rows, "goodput_rps") row["efficiency_pareto"] = not _is_dominated(row, rows, "goodput_per_accelerator") rows.sort(key=lambda r: (not r["pareto"], not r["slo_pass"], -r["goodput_rps"], r["p95_ttft_ms"])) return { "rows": rows, "pareto_count": sum(1 for r in rows if r["pareto"]), "efficiency_pareto_count": sum(1 for r in rows if r["efficiency_pareto"]), "candidate_count": len(rows), "objectives": [ "performance: maximize goodput / minimize p95 TTFT", "efficiency: maximize goodput per accelerator / minimize p95 TTFT", ], } def _design_row(result: dict, cfg: SimulationConfig, label: str) -> dict: resource = result["resource"] accelerator_instances = int(resource.get("accelerator_instances", 1)) goodput_rps = result["summary"]["goodput_rps"] return { "label": label, "topology": cfg.topology, "scheduler": cfg.scheduler, "max_batch_size": cfg.max_batch_size, "prefix_cache": cfg.prefix_cache_enabled, "accelerator_instances": accelerator_instances, "goodput_rps": goodput_rps, "goodput_per_accelerator": goodput_rps / max(accelerator_instances, 1), "slo_attainment": result["summary"]["slo_attainment"], "slo_pass": result["summary"]["slo_attainment"] >= cfg.slo_attainment_target and result["summary"]["requests_unfinished"] == 0, "p95_ttft_ms": result["latency"]["ttft_ms"]["p95"], "p95_e2e_ms": result["latency"]["e2e_ms"]["p95"], "peak_kv_utilization": resource.get("peak_kv_utilization", 0.0), "bottleneck": result["diagnostics"]["label"], }