diff --git a/packages/node/meshnet_node/performance_contract.py b/packages/node/meshnet_node/performance_contract.py new file mode 100644 index 0000000..5528b90 --- /dev/null +++ b/packages/node/meshnet_node/performance_contract.py @@ -0,0 +1,519 @@ +"""The versioned safetensors-versus-GGUF performance contract. + +The contract is the decision rule the native GGUF track is judged by, written +down *before* the numbers arrive and consumed later by the release gate +(DGR-014). Its thresholds are ratios against the Transformers/safetensors +reference recipe rather than absolute tokens/sec, because the absolute figure is +a property of whichever machine ran the benchmark and would have to be re-argued +on every host; a ratio is a claim about the runtime. + +Three rules give the contract its teeth: + +* **Thresholds are locked.** ``CONTRACT_SCHEMA_VERSION`` and ``locked_at`` + travel with the document. Moving a threshold after seeing results is a new + contract version and a human decision, not a tweak. +* **Only like-for-like comparisons count.** A recipe measured on a different + device than the reference is marked non-comparable and is granted no benefit, + so a GPU-versus-CPU mismatch can never be laundered into a speed win. +* **Quantized recipes never claim numerical equivalence.** Quality is gated on + the near-lossless quality lane; the performance-fit lane is judged on speed, + memory and fit alone. + +The verdict is one of ``promote``, ``optimize`` or ``stop`` — the three outcomes +the release gate is allowed to reach. +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Mapping + +from .recipe_benchmark import Lane + +# Layout of the contract document understood by this reader. +CONTRACT_SCHEMA_VERSION = 1 + +VERDICT_PROMOTE = "promote" +VERDICT_OPTIMIZE = "optimize" +VERDICT_STOP = "stop" + + +class PerformanceContractError(ValueError): + """Raised when a contract is missing, malformed, or of an unsupported version.""" + + +@dataclass(frozen=True) +class ContractThresholds: + """The locked decision thresholds. + + Every value is a ratio of a GGUF recipe's metric to the reference recipe's + metric on the same machine, same device, same plan. + + A *meaningful speed benefit* means the GGUF recipe decodes at least 25% + faster for a single request without making time-to-first-token materially + worse, or sustains at least 25% more aggregate throughput under concurrency. + Either route is a real win for the product: one helps a single user, the + other helps a loaded node. + + A *meaningful fit benefit* means peak resident memory (RSS plus VRAM) drops + by at least 25%. Fit is the product thesis — models larger than one + consumer node — so it is measured in resident bytes, not in how small the + file on disk is. Artifact size has its own reported threshold because a + smaller download is a real but secondary good. + + 25% is chosen to sit well clear of ordinary run-to-run variance on a busy + developer machine while still being a benefit a user would notice. A 5% + edge would not justify owning a native runtime and a patch stack. + """ + + min_decode_speedup: float = 1.25 + max_ttft_ratio: float = 1.25 + min_aggregate_throughput_speedup: float = 1.25 + max_resident_memory_ratio: float = 0.75 + max_artifact_size_ratio: float = 0.60 + min_quality_exact_match_rate: float = 0.90 + min_quality_mean_similarity: float = 0.97 + max_failure_rate: float = 0.0 + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass(frozen=True) +class PerformanceContract: + """A locked, versioned contract plus the baseline it was locked against.""" + + contract_version: int + locked_at: str + locked_by: str + plan_id: str + thresholds: ContractThresholds + baseline: Mapping[str, Any] + stop_condition: str + notes: str = "" + schema_version: int = CONTRACT_SCHEMA_VERSION + + def to_dict(self) -> dict: + return { + "schema_version": self.schema_version, + "contract_version": self.contract_version, + "locked_at": self.locked_at, + "locked_by": self.locked_by, + "plan_id": self.plan_id, + "thresholds": self.thresholds.to_dict(), + "baseline": dict(self.baseline), + "stop_condition": self.stop_condition, + "notes": self.notes, + } + + +STOP_CONDITION = ( + "Stop the native llama.cpp/GGUF track when, on the same machine and device " + "as the Transformers/safetensors reference and under this plan, no " + "performance-fit GGUF recipe delivers either a meaningful speed benefit " + "(>=25% higher single-request decode tokens/sec without a >25% worse TTFT, " + "or >=25% higher aggregate throughput under concurrency) or a meaningful fit " + "benefit (>=25% lower peak resident memory), or when the near-lossless " + "quality lane fails, which indicates a broken runtime rather than a " + "quantization trade-off." +) + + +def _recipe_entries(report: Mapping[str, Any]) -> dict[str, Mapping[str, Any]]: + return {entry["recipe"]["id"]: entry for entry in report["recipes"]} + + +def _cell(entry: Mapping[str, Any], concurrency: int) -> Mapping[str, Any] | None: + return entry["concurrency"].get(str(concurrency)) + + +def _resident_bytes(cell: Mapping[str, Any]) -> int: + return int(cell["peak_rss_bytes"]) + int(cell["peak_vram_bytes"]) + + +def _ratio(value: float, reference: float) -> float: + """Ratio guarded against a zero reference, which means "not measured".""" + if reference <= 0: + return 0.0 + return round(value / reference, 4) + + +@dataclass(frozen=True) +class RecipeEvaluation: + """How one GGUF recipe fared against the reference under the contract.""" + + recipe_id: str + lane: str + comparable: bool + incomparable_reason: str + speed_benefit: bool + fit_benefit: bool + quality_pass: bool | None + failures: int + measurements: Mapping[str, Any] + reasons: tuple[str, ...] + + def to_dict(self) -> dict: + data = asdict(self) + data["measurements"] = dict(self.measurements) + data["reasons"] = list(self.reasons) + return data + + +@dataclass(frozen=True) +class ContractEvaluation: + """The release-gate answer: a verdict plus every reason behind it.""" + + contract_version: int + plan_id: str + verdict: str + quality_lane_pass: bool + speed_benefit: bool + fit_benefit: bool + stop_condition_met: bool + recipes: tuple[RecipeEvaluation, ...] + rationale: tuple[str, ...] + + def to_dict(self) -> dict: + return { + "contract_version": self.contract_version, + "plan_id": self.plan_id, + "verdict": self.verdict, + "quality_lane_pass": self.quality_lane_pass, + "speed_benefit": self.speed_benefit, + "fit_benefit": self.fit_benefit, + "stop_condition_met": self.stop_condition_met, + "recipes": [recipe.to_dict() for recipe in self.recipes], + "rationale": list(self.rationale), + } + + +def _evaluate_recipe( + entry: Mapping[str, Any], + reference: Mapping[str, Any], + drift_by_recipe: Mapping[str, Mapping[str, Any]], + thresholds: ContractThresholds, + concurrency_levels: list[int], +) -> RecipeEvaluation: + recipe = entry["recipe"] + lane = recipe["lane"] + reasons: list[str] = [] + + if not entry["available"]: + return RecipeEvaluation( + recipe_id=recipe["id"], lane=lane, comparable=False, + incomparable_reason=entry["unavailable_reason"] or "recipe was not measured", + speed_benefit=False, fit_benefit=False, quality_pass=None, failures=0, + measurements={}, reasons=("recipe unavailable; no benefit granted",), + ) + + if recipe["device"] != reference["recipe"]["device"]: + return RecipeEvaluation( + recipe_id=recipe["id"], lane=lane, comparable=False, + incomparable_reason=( + f"recipe ran on device {recipe['device']!r} but the reference ran on " + f"{reference['recipe']['device']!r}; a cross-device ratio is not a runtime result" + ), + speed_benefit=False, fit_benefit=False, quality_pass=None, failures=0, + measurements={}, reasons=("cross-device comparison; no benefit granted",), + ) + + single = _cell(entry, 1) + reference_single = _cell(reference, 1) + measurements: dict[str, Any] = {} + speed_benefit = False + + if single and reference_single: + decode_speedup = _ratio( + single["decode_tokens_per_sec"], reference_single["decode_tokens_per_sec"] + ) + ttft_ratio = _ratio(single["ttft_p50_ms"], reference_single["ttft_p50_ms"]) + measurements["decode_speedup"] = decode_speedup + measurements["ttft_ratio"] = ttft_ratio + single_request_win = ( + decode_speedup >= thresholds.min_decode_speedup + and 0 < ttft_ratio <= thresholds.max_ttft_ratio + ) + if single_request_win: + speed_benefit = True + reasons.append( + f"single-request decode {decode_speedup:.2f}x reference " + f"(>= {thresholds.min_decode_speedup:.2f}x) at TTFT ratio {ttft_ratio:.2f}" + ) + else: + reasons.append( + f"no single-request speed win: decode {decode_speedup:.2f}x, TTFT {ttft_ratio:.2f}x" + ) + + concurrent = [level for level in concurrency_levels if level > 1] + if concurrent: + top = max(concurrent) + cell, reference_cell = _cell(entry, top), _cell(reference, top) + if cell and reference_cell: + aggregate_speedup = _ratio( + cell["aggregate_decode_tokens_per_sec"], + reference_cell["aggregate_decode_tokens_per_sec"], + ) + measurements["aggregate_throughput_speedup"] = aggregate_speedup + measurements["aggregate_concurrency"] = top + if aggregate_speedup >= thresholds.min_aggregate_throughput_speedup: + speed_benefit = True + reasons.append( + f"aggregate throughput at concurrency {top} is {aggregate_speedup:.2f}x reference " + f"(>= {thresholds.min_aggregate_throughput_speedup:.2f}x)" + ) + else: + reasons.append( + f"no concurrency speed win: aggregate throughput at {top} is " + f"{aggregate_speedup:.2f}x reference" + ) + + fit_benefit = False + if single and reference_single: + resident_ratio = _ratio(_resident_bytes(single), _resident_bytes(reference_single)) + artifact_ratio = _ratio( + entry["load"]["artifact_bytes"], reference["load"]["artifact_bytes"] + ) + measurements["resident_memory_ratio"] = resident_ratio + measurements["artifact_size_ratio"] = artifact_ratio + if 0 < resident_ratio <= thresholds.max_resident_memory_ratio: + fit_benefit = True + reasons.append( + f"peak resident memory is {resident_ratio:.2f}x reference " + f"(<= {thresholds.max_resident_memory_ratio:.2f}x)" + ) + else: + reasons.append(f"no fit win: peak resident memory is {resident_ratio:.2f}x reference") + measurements["artifact_size_win"] = ( + 0 < artifact_ratio <= thresholds.max_artifact_size_ratio + ) + + failures = sum(cell["failures"] for cell in entry["concurrency"].values()) + requests = sum(cell["requests"] for cell in entry["concurrency"].values()) + failure_rate = _ratio(failures, requests) if requests else 0.0 + measurements["failure_rate"] = failure_rate + if failure_rate > thresholds.max_failure_rate: + reasons.append(f"failure rate {failure_rate:.2%} exceeds the contract limit") + speed_benefit = False + fit_benefit = False + + # Quality is a claim only the near-lossless lane is allowed to make. A + # quantized recipe's drift is recorded elsewhere and deliberately not read + # here: Q4 disagreeing with bf16 is the trade-off, not a failure. + quality_pass: bool | None = None + if lane == Lane.QUALITY.value: + drift = drift_by_recipe.get(recipe["id"]) + if drift is None: + quality_pass = False + reasons.append("quality-lane recipe has no drift measurement against the reference") + else: + quality_pass = ( + drift["exact_match_rate"] >= thresholds.min_quality_exact_match_rate + and drift["mean_similarity"] >= thresholds.min_quality_mean_similarity + ) + measurements["exact_match_rate"] = drift["exact_match_rate"] + measurements["mean_similarity"] = drift["mean_similarity"] + reasons.append( + f"quality lane exact-match {drift['exact_match_rate']:.2f} / similarity " + f"{drift['mean_similarity']:.3f} versus the reference " + f"({'pass' if quality_pass else 'fail'})" + ) + + return RecipeEvaluation( + recipe_id=recipe["id"], lane=lane, comparable=True, incomparable_reason="", + speed_benefit=speed_benefit, fit_benefit=fit_benefit, quality_pass=quality_pass, + failures=failures, measurements=measurements, reasons=tuple(reasons), + ) + + +def evaluate_contract( + contract: PerformanceContract, + report: Mapping[str, Any], +) -> ContractEvaluation: + """Judge a benchmark report against the locked contract. + + Only performance-fit recipes can earn a speed or fit benefit; the quality + lane decides only whether the GGUF runtime is numerically sane. A GGUF + runtime that fails the quality lane is broken, and no amount of speed + redeems it, so the verdict is ``stop`` regardless of the other numbers. + """ + entries = _recipe_entries(report) + reference_id = report["reference_recipe_id"] + reference = entries.get(reference_id) + if reference is None: + raise PerformanceContractError( + f"report names reference recipe {reference_id!r}, which it does not contain" + ) + + drift_by_recipe = {entry["recipe_id"]: entry for entry in report["drift"]} + concurrency_levels = list(report["plan"]["concurrency_levels"]) + + evaluations = tuple( + _evaluate_recipe(entry, reference, drift_by_recipe, contract.thresholds, concurrency_levels) + for recipe_id, entry in entries.items() + if recipe_id != reference_id + ) + + quality_lane = [ + evaluation for evaluation in evaluations + if evaluation.lane == Lane.QUALITY.value and evaluation.comparable + ] + quality_lane_pass = bool(quality_lane) and all( + evaluation.quality_pass for evaluation in quality_lane + ) + performance_lane = [ + evaluation for evaluation in evaluations + if evaluation.lane == Lane.PERFORMANCE_FIT.value + ] + speed_benefit = any(evaluation.speed_benefit for evaluation in performance_lane) + fit_benefit = any(evaluation.fit_benefit for evaluation in performance_lane) + + rationale: list[str] = [] + if not quality_lane: + rationale.append( + "no comparable near-lossless GGUF recipe was measured, so the runtime's " + "numerical correctness is unproven" + ) + elif not quality_lane_pass: + rationale.append( + "the near-lossless quality lane failed: the GGUF runtime disagrees with the " + "safetensors reference beyond what near-lossless weights can explain" + ) + else: + rationale.append("the near-lossless quality lane passed against the safetensors reference") + + rationale.append( + "a meaningful speed benefit was measured" if speed_benefit + else "no performance-fit recipe delivered a meaningful speed benefit" + ) + rationale.append( + "a meaningful fit benefit was measured" if fit_benefit + else "no performance-fit recipe delivered a meaningful fit benefit" + ) + + stop_condition_met = not quality_lane_pass or not (speed_benefit or fit_benefit) + if stop_condition_met: + verdict = VERDICT_STOP + elif speed_benefit and fit_benefit: + verdict = VERDICT_PROMOTE + else: + verdict = VERDICT_OPTIMIZE + rationale.append( + "exactly one of speed or fit cleared the contract: the benefit is real but partial, " + "so the measured bottleneck needs a bounded optimization task before promotion" + ) + + return ContractEvaluation( + contract_version=contract.contract_version, + plan_id=contract.plan_id, + verdict=verdict, + quality_lane_pass=quality_lane_pass, + speed_benefit=speed_benefit, + fit_benefit=fit_benefit, + stop_condition_met=stop_condition_met, + recipes=evaluations, + rationale=tuple(rationale), + ) + + +def parse_contract(data: Any, source: str = "") -> PerformanceContract: + """Validate an already-decoded contract document.""" + if not isinstance(data, Mapping): + raise PerformanceContractError(f"contract root in {source} must be a JSON object") + + schema_version = data.get("schema_version") + if not isinstance(schema_version, int) or isinstance(schema_version, bool): + raise PerformanceContractError(f"'schema_version' in {source} must be an integer") + if schema_version != CONTRACT_SCHEMA_VERSION: + raise PerformanceContractError( + f"{source} declares contract schema version {schema_version}, but this node reads " + f"version {CONTRACT_SCHEMA_VERSION}; upgrade the node or use a supported contract" + ) + + for required in ("contract_version", "locked_at", "locked_by", "plan_id", "stop_condition"): + if not data.get(required): + raise PerformanceContractError(f"{source} is missing {required!r}") + + raw_thresholds = data.get("thresholds") + if not isinstance(raw_thresholds, Mapping): + raise PerformanceContractError(f"'thresholds' in {source} must be a JSON object") + known = {field for field in ContractThresholds().to_dict()} + unknown = set(raw_thresholds) - known + if unknown: + raise PerformanceContractError( + f"{source} carries unknown thresholds {sorted(unknown)}; this node enforces {sorted(known)}" + ) + thresholds = ContractThresholds(**{ + key: float(value) for key, value in raw_thresholds.items() + }) + + return PerformanceContract( + contract_version=int(data["contract_version"]), + locked_at=str(data["locked_at"]), + locked_by=str(data["locked_by"]), + plan_id=str(data["plan_id"]), + thresholds=thresholds, + baseline=dict(data.get("baseline") or {}), + stop_condition=str(data["stop_condition"]), + notes=str(data.get("notes", "")), + schema_version=schema_version, + ) + + +def load_contract(path: Path) -> PerformanceContract: + """Load and validate the contract at ``path``.""" + try: + raw = path.read_text(encoding="utf-8") + except OSError as exc: + raise PerformanceContractError( + f"cannot read performance contract {path}: {exc.strerror or exc}" + ) from exc + try: + data = json.loads(raw) + except json.JSONDecodeError as exc: + raise PerformanceContractError( + f"{path} is not valid JSON: {exc.msg} at line {exc.lineno} column {exc.colno}" + ) from exc + return parse_contract(data, source=str(path)) + + +def baseline_from_report(report: Mapping[str, Any]) -> dict[str, Any]: + """Distil the reference numbers a later gate needs to compare against.""" + entries = _recipe_entries(report) + baseline: dict[str, Any] = { + "evidence_class": report["evidence_class"], + "model_id": report["plan"]["model_id"], + "model_revision": report["plan"]["model_revision"], + "reference_recipe_id": report["reference_recipe_id"], + "host": report["host"], + "recipes": {}, + } + for recipe_id, entry in entries.items(): + if not entry["available"]: + baseline["recipes"][recipe_id] = {"available": False, + "reason": entry["unavailable_reason"]} + continue + baseline["recipes"][recipe_id] = { + "available": True, + "lane": entry["recipe"]["lane"], + "device": entry["recipe"]["device"], + "artifact_bytes": entry["load"]["artifact_bytes"], + "concurrency": { + level: { + "ttft_p50_ms": cell["ttft_p50_ms"], + "ttft_p95_ms": cell["ttft_p95_ms"], + "latency_p50_ms": cell["latency_p50_ms"], + "latency_p95_ms": cell["latency_p95_ms"], + "prefill_tokens_per_sec": cell["prefill_tokens_per_sec"], + "decode_tokens_per_sec": cell["decode_tokens_per_sec"], + "aggregate_decode_tokens_per_sec": cell["aggregate_decode_tokens_per_sec"], + "peak_rss_bytes": cell["peak_rss_bytes"], + "peak_vram_bytes": cell["peak_vram_bytes"], + "failures": cell["failures"], + } + for level, cell in entry["concurrency"].items() + }, + } + return baseline diff --git a/packages/node/meshnet_node/recipe_benchmark.py b/packages/node/meshnet_node/recipe_benchmark.py new file mode 100644 index 0000000..9bfaf16 --- /dev/null +++ b/packages/node/meshnet_node/recipe_benchmark.py @@ -0,0 +1,648 @@ +"""Controlled safetensors-versus-GGUF recipe benchmark. + +This is a *model recipe* benchmark, unlike +:mod:`meshnet_node.route_session_benchmark`, which is a transport harness. It +answers one question: on one machine, with one model revision and one fixed +workload, what do the Transformers/safetensors recipe and the whole-model +llama.cpp/GGUF recipes actually cost in speed, memory, fit, and output drift? + +Two ideas keep the answer honest. + +**Lanes.** A recipe belongs to exactly one :class:`Lane`. The quality lane +holds near-lossless recipes (bf16/f16 weights) whose outputs may legitimately be +compared for numerical agreement. The performance-fit lane holds quantized +recipes (Q8_0, Q4_K_M, ...). Quantized recipes are judged on speed, memory, and +artifact size only; their drift is *reported* but never read as evidence that +Q4 and bf16 are numerically equivalent, because they are not. The lane is a +property of the recipe, so nothing downstream can quietly cross the boundary. + +**Drivers.** The measurement core here is pure and runtime-free: it drives a +:class:`RecipeDriver` and computes metrics. Real runtimes live in +:mod:`meshnet_node.recipe_drivers` and are imported only on demand, which keeps +the default test suite deterministic, GPU-free and model-download-free while the +same code path produces the real evidence. +""" + +from __future__ import annotations + +import argparse +import json +import statistics +import time +from concurrent.futures import ThreadPoolExecutor +from dataclasses import asdict, dataclass, field +from difflib import SequenceMatcher +from enum import Enum +from pathlib import Path +from typing import Any, Protocol, Sequence + +# Layout of the report document produced by :func:`build_report`. +REPORT_SCHEMA_VERSION = 1 + + +class Lane(str, Enum): + """Why a recipe is being measured at all. + + ``QUALITY`` recipes carry near-lossless weights, so comparing their output + with the reference recipe is a meaningful correctness signal. + ``PERFORMANCE_FIT`` recipes carry quantized weights: they exist to be faster + or to fit, and their drift is descriptive, never a pass/fail equivalence + claim. + """ + + QUALITY = "quality" + PERFORMANCE_FIT = "performance-fit" + + +class BenchmarkError(RuntimeError): + """Raised when a benchmark cannot be run as specified.""" + + +@dataclass(frozen=True) +class SamplingPolicy: + """The sampling policy every recipe must be given, identically. + + Greedy by default: sampling noise would otherwise be indistinguishable from + quantization drift, and the whole point of the quality lane is to tell those + two apart. + """ + + temperature: float = 0.0 + top_p: float = 1.0 + top_k: int = 1 + seed: int = 1234 + max_output_tokens: int = 64 + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass(frozen=True) +class PromptSpec: + """One fixed prompt, tagged with the context length it is meant to exercise.""" + + id: str + text: str + context_class: str = "short" + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass(frozen=True) +class BenchmarkPlan: + """The controlled variables: identical for every recipe in a report. + + A plan is the experiment. If two recipes were measured under different + plans, their numbers are not comparable and the report must not pretend they + are, so the plan is recorded once at the top of the document rather than + per-recipe. + """ + + plan_id: str + model_id: str + model_revision: str + prompts: tuple[PromptSpec, ...] + sampling: SamplingPolicy = SamplingPolicy() + concurrency_levels: tuple[int, ...] = (1, 4) + repeats: int = 1 + warmup_requests: int = 1 + + def __post_init__(self) -> None: + if not self.prompts: + raise BenchmarkError("a benchmark plan needs at least one prompt") + if not self.concurrency_levels or any(level < 1 for level in self.concurrency_levels): + raise BenchmarkError("concurrency levels must all be >= 1") + if self.repeats < 1: + raise BenchmarkError("repeats must be >= 1") + + def to_dict(self) -> dict: + return { + "plan_id": self.plan_id, + "model_id": self.model_id, + "model_revision": self.model_revision, + "prompts": [prompt.to_dict() for prompt in self.prompts], + "sampling": self.sampling.to_dict(), + "concurrency_levels": list(self.concurrency_levels), + "repeats": self.repeats, + "warmup_requests": self.warmup_requests, + } + + +@dataclass(frozen=True) +class RecipeSpec: + """One runtime recipe under test. + + ``is_reference`` marks the single recipe every other recipe's output drift is + measured against — the current Transformers/safetensors route, which + decision Gate 8 keeps as the correctness backend. + """ + + id: str + runtime: str + weight_format: str + weight_quantization: str + lane: Lane + device: str + artifact_path: str = "" + is_reference: bool = False + notes: str = "" + + def to_dict(self) -> dict: + data = asdict(self) + data["lane"] = self.lane.value + return data + + +@dataclass(frozen=True) +class LoadStats: + """What loading the recipe cost, before any token is generated.""" + + artifact_bytes: int + load_ms: float + rss_bytes: int = 0 + vram_bytes: int = 0 + backend_detail: str = "" + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass(frozen=True) +class GenerationSample: + """One completed generation as reported by a driver. + + ``prefill_ms``/``decode_ms`` are the runtime's own split where it exposes one + (llama.cpp does); drivers that cannot split honestly report ``prefill_ms`` as + the time to the first token. ``queue_wait_ms`` separates time spent waiting + for a runtime slot from time spent computing, so a concurrency-4 TTFT is not + silently read as a slower prefill. + """ + + text: str + prompt_tokens: int + decode_tokens: int + ttft_ms: float + prefill_ms: float + decode_ms: float + total_ms: float + queue_wait_ms: float = 0.0 + + +class RecipeDriver(Protocol): + """The seam every runtime implements; the measurement core knows nothing else.""" + + def load(self) -> LoadStats: + """Load the artifact and return its cost.""" + + def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample: + """Run one complete generation under the given sampling policy.""" + + def memory_probe(self) -> tuple[int, int]: + """Return ``(rss_bytes, vram_bytes)`` observed right now.""" + + def close(self) -> None: + """Release the runtime.""" + + +@dataclass(frozen=True) +class RequestOutcome: + """One request attempt, successful or not. + + A failure is a first-class result, not an exception that aborts the run: a + recipe that cannot sustain concurrency 4 has told us something, and the + report must carry it rather than lose it. + """ + + recipe_id: str + concurrency: int + prompt_id: str + repeat: int + ok: bool + latency_ms: float + ttft_ms: float = 0.0 + prefill_ms: float = 0.0 + decode_ms: float = 0.0 + queue_wait_ms: float = 0.0 + prompt_tokens: int = 0 + decode_tokens: int = 0 + text: str = "" + error: str = "" + + def to_dict(self) -> dict: + return asdict(self) + + +def _percentile(values: Sequence[float], percentile: float) -> float: + """Nearest-rank percentile; 0.0 for an empty sample.""" + ordered = sorted(values) + if not ordered: + return 0.0 + rank = max(1, -(-len(ordered) * percentile // 100)) + return round(ordered[int(rank) - 1], 4) + + +def _mean(values: Sequence[float]) -> float: + return round(statistics.fmean(values), 4) if values else 0.0 + + +@dataclass(frozen=True) +class ConcurrencyMetrics: + """Aggregate metrics for one recipe at one concurrency level.""" + + concurrency: int + requests: int + failures: int + wall_ms: float + ttft_p50_ms: float + ttft_p95_ms: float + latency_p50_ms: float + latency_p95_ms: float + prefill_tokens_per_sec: float + decode_tokens_per_sec: float + aggregate_decode_tokens_per_sec: float + peak_rss_bytes: int + peak_vram_bytes: int + failure_reasons: tuple[str, ...] = () + + def to_dict(self) -> dict: + data = asdict(self) + data["failure_reasons"] = list(self.failure_reasons) + return data + + +def summarize_concurrency( + outcomes: Sequence[RequestOutcome], + *, + concurrency: int, + wall_ms: float, + peak_rss_bytes: int, + peak_vram_bytes: int, +) -> ConcurrencyMetrics: + """Aggregate one recipe/concurrency cell. + + Per-request rates are averaged over successful requests; aggregate + throughput is total decoded tokens over the wall clock of the whole cell, + which is the only figure that credits a runtime for overlapping work. + """ + ok = [outcome for outcome in outcomes if outcome.ok] + failures = [outcome for outcome in outcomes if not outcome.ok] + decode_tokens = sum(outcome.decode_tokens for outcome in ok) + + prefill_rates = [ + outcome.prompt_tokens / (outcome.prefill_ms / 1000) + for outcome in ok + if outcome.prefill_ms > 0 and outcome.prompt_tokens + ] + decode_rates = [ + outcome.decode_tokens / (outcome.decode_ms / 1000) + for outcome in ok + if outcome.decode_ms > 0 and outcome.decode_tokens + ] + return ConcurrencyMetrics( + concurrency=concurrency, + requests=len(outcomes), + failures=len(failures), + wall_ms=round(wall_ms, 4), + ttft_p50_ms=_percentile([outcome.ttft_ms for outcome in ok], 50), + ttft_p95_ms=_percentile([outcome.ttft_ms for outcome in ok], 95), + latency_p50_ms=_percentile([outcome.latency_ms for outcome in ok], 50), + latency_p95_ms=_percentile([outcome.latency_ms for outcome in ok], 95), + prefill_tokens_per_sec=_mean(prefill_rates), + decode_tokens_per_sec=_mean(decode_rates), + aggregate_decode_tokens_per_sec=round(decode_tokens / max(1e-6, wall_ms / 1000), 4), + peak_rss_bytes=peak_rss_bytes, + peak_vram_bytes=peak_vram_bytes, + failure_reasons=tuple(sorted({outcome.error for outcome in failures if outcome.error})), + ) + + +@dataclass +class RecipeMeasurement: + """Everything measured for one recipe across every concurrency level.""" + + recipe: RecipeSpec + load: LoadStats + metrics: dict[int, ConcurrencyMetrics] = field(default_factory=dict) + outcomes: list[RequestOutcome] = field(default_factory=list) + unavailable_reason: str = "" + + @property + def available(self) -> bool: + return not self.unavailable_reason + + def outputs_by_prompt(self) -> dict[str, str]: + """First successful output per prompt, at the lowest concurrency measured. + + Drift is a property of the recipe, not of load: concurrency must not + change greedy output, so the least-contended sample is the fair one. + """ + best: dict[str, tuple[int, str]] = {} + for outcome in self.outcomes: + if not outcome.ok: + continue + seen = best.get(outcome.prompt_id) + if seen is None or outcome.concurrency < seen[0]: + best[outcome.prompt_id] = (outcome.concurrency, outcome.text) + return {prompt_id: text for prompt_id, (_, text) in best.items()} + + def to_dict(self) -> dict: + return { + "recipe": self.recipe.to_dict(), + "available": self.available, + "unavailable_reason": self.unavailable_reason, + "load": self.load.to_dict(), + "concurrency": { + str(level): metrics.to_dict() for level, metrics in sorted(self.metrics.items()) + }, + "outcomes": [outcome.to_dict() for outcome in self.outcomes], + } + + +@dataclass(frozen=True) +class DriftReport: + """Output drift of one recipe against the reference recipe. + + ``advisory`` is true for every performance-fit recipe: the number is + published, but a Q4 recipe disagreeing with bf16 is expected behaviour, not a + defect, and no gate may read it as one. + """ + + recipe_id: str + lane: Lane + reference_id: str + compared_prompts: int + exact_match_rate: float + mean_similarity: float + advisory: bool + per_prompt: tuple[dict[str, Any], ...] = () + + def to_dict(self) -> dict: + return { + "recipe_id": self.recipe_id, + "lane": self.lane.value, + "reference_id": self.reference_id, + "compared_prompts": self.compared_prompts, + "exact_match_rate": self.exact_match_rate, + "mean_similarity": self.mean_similarity, + "advisory": self.advisory, + "per_prompt": list(self.per_prompt), + } + + +def _first_divergence(left: str, right: str) -> int: + """Index of the first differing character, or -1 when the strings agree.""" + for index, (a, b) in enumerate(zip(left, right)): + if a != b: + return index + return -1 if len(left) == len(right) else min(len(left), len(right)) + + +def compute_drift( + measurement: RecipeMeasurement, + reference: RecipeMeasurement, +) -> DriftReport: + """Compare one recipe's greedy outputs with the reference recipe's.""" + reference_outputs = reference.outputs_by_prompt() + outputs = measurement.outputs_by_prompt() + shared = sorted(set(outputs) & set(reference_outputs)) + + per_prompt: list[dict[str, Any]] = [] + exact = 0 + similarities: list[float] = [] + for prompt_id in shared: + got, want = outputs[prompt_id], reference_outputs[prompt_id] + matches = got == want + exact += matches + similarity = round(SequenceMatcher(None, want, got).ratio(), 4) + similarities.append(similarity) + per_prompt.append({ + "prompt_id": prompt_id, + "exact_match": matches, + "similarity": similarity, + "first_divergence_char": _first_divergence(want, got), + "reference_text": want, + "recipe_text": got, + }) + + return DriftReport( + recipe_id=measurement.recipe.id, + lane=measurement.recipe.lane, + reference_id=reference.recipe.id, + compared_prompts=len(shared), + exact_match_rate=round(exact / len(shared), 4) if shared else 0.0, + mean_similarity=_mean(similarities), + advisory=measurement.recipe.lane is Lane.PERFORMANCE_FIT, + per_prompt=tuple(per_prompt), + ) + + +class _PeakMemory: + """Sample a driver's memory while requests are in flight.""" + + def __init__(self, driver: RecipeDriver) -> None: + self._driver = driver + self.peak_rss = 0 + self.peak_vram = 0 + + def sample(self) -> None: + try: + rss, vram = self._driver.memory_probe() + except Exception: # a probe must never fail a benchmark run + return + self.peak_rss = max(self.peak_rss, rss) + self.peak_vram = max(self.peak_vram, vram) + + +def _run_request( + driver: RecipeDriver, + recipe: RecipeSpec, + prompt: PromptSpec, + sampling: SamplingPolicy, + concurrency: int, + repeat: int, + memory: _PeakMemory, +) -> RequestOutcome: + started = time.monotonic() + try: + sample = driver.generate(prompt.text, sampling) + except Exception as exc: # a failed request is data, not a crashed benchmark + return RequestOutcome( + recipe_id=recipe.id, + concurrency=concurrency, + prompt_id=prompt.id, + repeat=repeat, + ok=False, + latency_ms=round((time.monotonic() - started) * 1000, 4), + error=f"{type(exc).__name__}: {exc}", + ) + finally: + memory.sample() + + return RequestOutcome( + recipe_id=recipe.id, + concurrency=concurrency, + prompt_id=prompt.id, + repeat=repeat, + ok=True, + latency_ms=round(sample.total_ms, 4), + ttft_ms=round(sample.ttft_ms, 4), + prefill_ms=round(sample.prefill_ms, 4), + decode_ms=round(sample.decode_ms, 4), + queue_wait_ms=round(sample.queue_wait_ms, 4), + prompt_tokens=sample.prompt_tokens, + decode_tokens=sample.decode_tokens, + text=sample.text, + ) + + +def measure_recipe( + driver: RecipeDriver, + recipe: RecipeSpec, + plan: BenchmarkPlan, +) -> RecipeMeasurement: + """Load one recipe and run the whole plan against it. + + The driver is closed exactly once, whatever happens, so a recipe that dies at + concurrency 4 still releases its weights before the next recipe loads. + """ + load = driver.load() + memory = _PeakMemory(driver) + memory.sample() + measurement = RecipeMeasurement(recipe=recipe, load=load) + + try: + for _ in range(plan.warmup_requests): + try: + driver.generate(plan.prompts[0].text, plan.sampling) + except Exception: # a failing warmup is reported by the real requests + break + + for concurrency in plan.concurrency_levels: + requests = [ + (prompt, repeat) + for repeat in range(plan.repeats) + for prompt in plan.prompts + for _ in range(concurrency) + ] + started = time.monotonic() + with ThreadPoolExecutor(max_workers=concurrency) as pool: + outcomes = list(pool.map( + lambda item: _run_request( + driver, recipe, item[0], plan.sampling, concurrency, item[1], memory + ), + requests, + )) + wall_ms = (time.monotonic() - started) * 1000 + + measurement.outcomes.extend(outcomes) + measurement.metrics[concurrency] = summarize_concurrency( + outcomes, + concurrency=concurrency, + wall_ms=wall_ms, + peak_rss_bytes=memory.peak_rss, + peak_vram_bytes=memory.peak_vram, + ) + finally: + driver.close() + + return measurement + + +def build_report( + plan: BenchmarkPlan, + measurements: Sequence[RecipeMeasurement], + *, + host: dict[str, Any], + evidence_class: str, +) -> dict: + """Assemble the machine-readable benchmark document. + + ``evidence_class`` is one of ``synthetic``, ``local-real`` or + ``multi-machine-real`` and is never inferred: a report that cannot say how it + was produced cannot be trusted by a release gate. + """ + if evidence_class not in {"synthetic", "local-real", "multi-machine-real"}: + raise BenchmarkError(f"unknown evidence class {evidence_class!r}") + + references = [m for m in measurements if m.recipe.is_reference] + if len(references) != 1: + raise BenchmarkError( + f"exactly one reference recipe is required, got {len(references)}" + ) + reference = references[0] + if reference.recipe.lane is not Lane.QUALITY: + raise BenchmarkError("the reference recipe must sit in the quality lane") + + drift = [ + compute_drift(measurement, reference).to_dict() + for measurement in measurements + if measurement is not reference and measurement.available + ] + return { + "schema_version": REPORT_SCHEMA_VERSION, + "evidence_class": evidence_class, + "plan": plan.to_dict(), + "host": host, + "reference_recipe_id": reference.recipe.id, + "recipes": [measurement.to_dict() for measurement in measurements], + "drift": drift, + } + + +def format_summary(report: dict) -> str: + """Render the human-readable companion to the JSON artifact.""" + plan = report["plan"] + lines = [ + f"Recipe benchmark {plan['plan_id']} ({report['evidence_class']})", + f"model {plan['model_id']}@{plan['model_revision']}", + ] + for entry in report["recipes"]: + recipe = entry["recipe"] + if not entry["available"]: + lines.append(f"{recipe['id']:38} UNAVAILABLE: {entry['unavailable_reason']}") + continue + artifact_gb = entry["load"]["artifact_bytes"] / 1e9 + for level, metrics in entry["concurrency"].items(): + lines.append( + f"{recipe['id']:38} [{recipe['lane']:16}] c={level:>2} " + f"ttft p50/p95 {metrics['ttft_p50_ms']:8.1f}/{metrics['ttft_p95_ms']:8.1f} ms; " + f"prefill {metrics['prefill_tokens_per_sec']:7.1f} tok/s; " + f"decode {metrics['decode_tokens_per_sec']:6.1f} tok/s; " + f"aggregate {metrics['aggregate_decode_tokens_per_sec']:7.1f} tok/s; " + f"rss {metrics['peak_rss_bytes'] / 1e9:5.2f} GB; " + f"vram {metrics['peak_vram_bytes'] / 1e9:5.2f} GB; " + f"artifact {artifact_gb:5.2f} GB; failures {metrics['failures']}" + ) + for entry in report["drift"]: + tag = "advisory" if entry["advisory"] else "gated" + lines.append( + f"drift {entry['recipe_id']:32} vs {entry['reference_id']:28} " + f"exact {entry['exact_match_rate']:.2f}; similarity {entry['mean_similarity']:.3f} ({tag})" + ) + return "\n".join(lines) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Run the controlled safetensors-versus-GGUF recipe benchmark" + ) + parser.add_argument("--config", type=Path, required=True, help="benchmark configuration JSON") + parser.add_argument("--json-out", type=Path, help="write the JSON report to this path") + parser.add_argument("--summary-out", type=Path, help="write the text summary to this path") + args = parser.parse_args(argv) + + from .recipe_drivers import run_configured_benchmark # heavy runtimes: import on demand + + report = run_configured_benchmark(json.loads(args.config.read_text(encoding="utf-8"))) + summary = format_summary(report) + if args.json_out: + args.json_out.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") + if args.summary_out: + args.summary_out.write_text(summary + "\n", encoding="utf-8") + print(summary) + return 0 + + +if __name__ == "__main__": # pragma: no cover - CLI entry point + raise SystemExit(main()) diff --git a/packages/node/meshnet_node/recipe_drivers.py b/packages/node/meshnet_node/recipe_drivers.py new file mode 100644 index 0000000..0ce96e2 --- /dev/null +++ b/packages/node/meshnet_node/recipe_drivers.py @@ -0,0 +1,473 @@ +"""Real runtime drivers for the recipe benchmark. + +This module is the only place that imports torch, transformers, or spawns a +llama.cpp server, and :mod:`meshnet_node.recipe_benchmark` imports it lazily. +That keeps the default test suite deterministic, GPU-free and download-free +while the real evidence runs through exactly the same measurement core. + +Fairness is the whole point of a baseline, so both drivers are held to the same +rules: + +* They are handed a **pre-formatted prompt string**. Neither applies a chat + template, because a template applied twice — or differently — by two runtimes + would show up as a speed and drift difference that has nothing to do with the + runtime. +* They are given the **same CPU thread budget**, so the comparison measures + kernels rather than how many cores each runtime felt entitled to take. +* They report the runtime's **own prefill/decode split** where it has one, and + say so honestly where it does not. +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import time +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any, Mapping + +from .recipe_benchmark import ( + BenchmarkError, + BenchmarkPlan, + GenerationSample, + Lane, + LoadStats, + PromptSpec, + RecipeSpec, + SamplingPolicy, + build_report, + measure_recipe, +) + +REAL_INFERENCE_ENV = "MESHNET_ENABLE_REAL_INFERENCE_TESTS" + + +def real_inference_enabled() -> bool: + """Real runtimes stay off unless the operator opts in explicitly.""" + return os.environ.get(REAL_INFERENCE_ENV) == "1" + + +def require_real_inference() -> None: + if not real_inference_enabled(): + raise BenchmarkError( + f"real model execution is opt-in: set {REAL_INFERENCE_ENV}=1 to run this benchmark" + ) + + +def _process_rss(pid: int | None = None) -> int: + """Resident bytes for a process and its children, or 0 when unobservable.""" + try: + import psutil + except ImportError: + return 0 + try: + process = psutil.Process(pid) if pid else psutil.Process() + total = process.memory_info().rss + for child in process.children(recursive=True): + try: + total += child.memory_info().rss + except psutil.Error: + continue + return int(total) + except Exception: + return 0 + + +def _directory_bytes(path: Path) -> int: + if path.is_file(): + return path.stat().st_size + return sum(entry.stat().st_size for entry in path.rglob("*") if entry.is_file()) + + +class TransformersDriver: + """The current Transformers/safetensors recipe: the correctness reference. + + Generation is a hand-written prefill-then-decode loop rather than + ``model.generate`` because the benchmark needs the two phases separated: one + forward over the prompt gives an exact prefill time and TTFT, and the cached + single-token steps that follow give an exact decode rate. ``generate`` would + hand back one blended number. + """ + + def __init__( + self, + model_path: str, + *, + device: str = "cpu", + dtype: str = "bfloat16", + threads: int = 8, + ) -> None: + self.model_path = Path(model_path) + self.device = device + self.dtype = dtype + self.threads = threads + self._model: Any = None + self._tokenizer: Any = None + self._torch: Any = None + + def load(self) -> LoadStats: + import torch + from transformers import AutoModelForCausalLM, AutoTokenizer + + self._torch = torch + torch.set_num_threads(self.threads) + torch.manual_seed(0) + + started = time.monotonic() + self._tokenizer = AutoTokenizer.from_pretrained( + str(self.model_path), local_files_only=True + ) + self._model = AutoModelForCausalLM.from_pretrained( + str(self.model_path), + dtype=getattr(torch, self.dtype), + local_files_only=True, + ) + self._model.to(self.device) + self._model.eval() + load_ms = (time.monotonic() - started) * 1000 + + return LoadStats( + artifact_bytes=_directory_bytes(self.model_path), + load_ms=round(load_ms, 4), + rss_bytes=_process_rss(), + vram_bytes=self._vram_bytes(), + backend_detail=( + f"torch {torch.__version__}; dtype {self.dtype}; " + f"device {self.device}; intra-op threads {self.threads}" + ), + ) + + def _vram_bytes(self) -> int: + torch = self._torch + if torch is None or self.device == "cpu": + return 0 + try: + if torch.cuda.is_available(): + return int(torch.cuda.max_memory_allocated()) + except Exception: + return 0 + return 0 + + def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample: + if self._model is None: + raise BenchmarkError("TransformersDriver.generate called before load()") + torch = self._torch + + # add_special_tokens=False: the plan owns the prompt format, and the + # llama.cpp recipe is given the identical string. + encoded = self._tokenizer(prompt, return_tensors="pt", add_special_tokens=False) + input_ids = encoded["input_ids"].to(self.device) + prompt_tokens = int(input_ids.shape[-1]) + eos_ids = {self._tokenizer.eos_token_id} | set( + getattr(self._model.generation_config, "eos_token_id", None) or [] + if isinstance(getattr(self._model.generation_config, "eos_token_id", None), list) + else [] + ) + eos_ids.discard(None) + + started = time.monotonic() + with torch.inference_mode(): + outputs = self._model(input_ids=input_ids, use_cache=True) + past = outputs.past_key_values + next_id = self._select(outputs.logits[:, -1, :], sampling) + ttft_ms = (time.monotonic() - started) * 1000 + + token_ids = [int(next_id.item())] + decode_started = time.monotonic() + while len(token_ids) < sampling.max_output_tokens and token_ids[-1] not in eos_ids: + outputs = self._model( + input_ids=next_id.view(1, 1), past_key_values=past, use_cache=True + ) + past = outputs.past_key_values + next_id = self._select(outputs.logits[:, -1, :], sampling) + token_ids.append(int(next_id.item())) + decode_ms = (time.monotonic() - decode_started) * 1000 + + total_ms = (time.monotonic() - started) * 1000 + emitted = [token for token in token_ids if token not in eos_ids] + return GenerationSample( + text=self._tokenizer.decode(emitted, skip_special_tokens=True), + prompt_tokens=prompt_tokens, + # The first token is produced by the prefill forward, so the decode + # rate must not be credited with it. + decode_tokens=max(0, len(token_ids) - 1), + ttft_ms=ttft_ms, + prefill_ms=ttft_ms, + decode_ms=decode_ms, + total_ms=total_ms, + ) + + def _select(self, logits: Any, sampling: SamplingPolicy) -> Any: + if sampling.temperature > 0: + raise BenchmarkError( + "this benchmark is greedy-only: sampling noise is indistinguishable from " + "quantization drift, which is precisely what the quality lane must isolate" + ) + return logits.argmax(dim=-1) + + def memory_probe(self) -> tuple[int, int]: + return _process_rss(), self._vram_bytes() + + def close(self) -> None: + self._model = None + self._tokenizer = None + if self._torch is not None: + import gc + + gc.collect() + + +def _free_port() -> int: + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + return int(probe.getsockname()[1]) + + +class LlamaCppServerDriver: + """The whole-model llama.cpp/GGUF recipe, driven through ``llama-server``. + + ``llama-server`` is used rather than an in-process binding because it is the + shape llama.cpp is actually deployed in and the only one that offers + continuous batching across parallel slots — which is the runtime property + this project cares about most. It also reports its own prefill/decode + timings per request, so the decode rate is the runtime's own number and not + an inference drawn from a client-side stopwatch. + """ + + def __init__( + self, + binary: str, + gguf_path: str, + *, + device: str = "cpu", + threads: int = 8, + n_parallel: int = 4, + context_per_slot: int = 1024, + n_gpu_layers: int = 0, + startup_timeout_s: float = 120.0, + ) -> None: + self.binary = Path(binary) + self.gguf_path = Path(gguf_path) + self.device = device + self.threads = threads + self.n_parallel = n_parallel + self.context_per_slot = context_per_slot + self.n_gpu_layers = n_gpu_layers + self.startup_timeout_s = startup_timeout_s + self._process: subprocess.Popen | None = None + self._port = 0 + self._log: Any = None + + @property + def _url(self) -> str: + return f"http://127.0.0.1:{self._port}" + + def load(self) -> LoadStats: + if not self.binary.exists(): + raise BenchmarkError(f"llama-server binary not found at {self.binary}") + if not self.gguf_path.exists(): + raise BenchmarkError(f"GGUF artifact not found at {self.gguf_path}") + + self._port = _free_port() + command = [ + str(self.binary), + "--model", str(self.gguf_path), + "--host", "127.0.0.1", + "--port", str(self._port), + "--threads", str(self.threads), + "--parallel", str(self.n_parallel), + # Every slot must hold a whole request, so the pool is sized for the + # worst case rather than letting llama.cpp silently truncate context. + "--ctx-size", str(self.context_per_slot * self.n_parallel), + "--n-gpu-layers", str(self.n_gpu_layers), + "--no-webui", + ] + started = time.monotonic() + self._log = subprocess.PIPE + self._process = subprocess.Popen( + command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + ) + self._await_health(started) + load_ms = (time.monotonic() - started) * 1000 + + return LoadStats( + artifact_bytes=self.gguf_path.stat().st_size, + load_ms=round(load_ms, 4), + rss_bytes=_process_rss(self._process.pid), + vram_bytes=0, + backend_detail=( + f"llama-server; threads {self.threads}; parallel slots {self.n_parallel}; " + f"ctx/slot {self.context_per_slot}; gpu layers {self.n_gpu_layers}" + ), + ) + + def _await_health(self, started: float) -> None: + while time.monotonic() - started < self.startup_timeout_s: + if self._process is not None and self._process.poll() is not None: + raise BenchmarkError( + f"llama-server exited with code {self._process.returncode} during startup" + ) + try: + with urllib.request.urlopen(f"{self._url}/health", timeout=2) as response: + if response.status == 200: + return + except (urllib.error.URLError, OSError): + time.sleep(0.25) + raise BenchmarkError( + f"llama-server did not become healthy within {self.startup_timeout_s:.0f}s" + ) + + def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample: + if self._process is None: + raise BenchmarkError("LlamaCppServerDriver.generate called before load()") + if sampling.temperature > 0: + raise BenchmarkError("this benchmark is greedy-only; see TransformersDriver._select") + + body = json.dumps({ + "prompt": prompt, + "n_predict": sampling.max_output_tokens, + "temperature": 0.0, + "top_k": 1, + "top_p": 1.0, + "seed": sampling.seed, + # Prompt cache reuse across repeats would measure the cache, not the + # prefill, and the safetensors recipe has no equivalent. + "cache_prompt": False, + "stream": True, + }).encode() + request = urllib.request.Request( + f"{self._url}/completion", data=body, + headers={"Content-Type": "application/json"}, method="POST", + ) + + started = time.monotonic() + ttft_ms = 0.0 + chunks: list[str] = [] + timings: Mapping[str, Any] = {} + with urllib.request.urlopen(request, timeout=600) as response: + for raw in response: + line = raw.decode("utf-8").strip() + if not line.startswith("data:"): + continue + payload = json.loads(line[len("data:"):].strip()) + content = payload.get("content", "") + if content and not ttft_ms: + ttft_ms = (time.monotonic() - started) * 1000 + chunks.append(content) + if payload.get("stop"): + timings = payload.get("timings") or {} + total_ms = (time.monotonic() - started) * 1000 + + if not timings: + raise BenchmarkError("llama-server returned no timings; cannot report an honest split") + + prefill_ms = float(timings.get("prompt_ms", 0.0)) + decode_ms = float(timings.get("predicted_ms", 0.0)) + return GenerationSample( + text="".join(chunks), + prompt_tokens=int(timings.get("prompt_n", 0)), + decode_tokens=int(timings.get("predicted_n", 0)), + ttft_ms=ttft_ms or total_ms, + prefill_ms=prefill_ms, + decode_ms=decode_ms, + total_ms=total_ms, + # Whatever the wall clock saw but the runtime did not attribute to + # compute is time this request spent waiting for a slot. + queue_wait_ms=max(0.0, total_ms - prefill_ms - decode_ms), + ) + + def memory_probe(self) -> tuple[int, int]: + if self._process is None: + return 0, 0 + return _process_rss(self._process.pid), 0 + + def close(self) -> None: + if self._process is None: + return + self._process.terminate() + try: + self._process.wait(timeout=20) + except subprocess.TimeoutExpired: + self._process.kill() + self._process.wait(timeout=10) + self._process = None + + +def build_driver(spec: Mapping[str, Any], plan: BenchmarkPlan) -> RecipeDriverBundle: + """Construct the driver named by a recipe's ``driver`` block.""" + driver_spec = dict(spec["driver"]) + kind = driver_spec.pop("type") + if kind == "transformers": + return TransformersDriver(**driver_spec) + if kind == "llama-cpp-server": + driver_spec.setdefault("n_parallel", max(plan.concurrency_levels)) + return LlamaCppServerDriver(**driver_spec) + raise BenchmarkError(f"unknown driver type {kind!r}") + + +RecipeDriverBundle = Any # a RecipeDriver; named for readability at the call site + + +def _plan_from_config(config: Mapping[str, Any]) -> BenchmarkPlan: + raw = config["plan"] + return BenchmarkPlan( + plan_id=raw["plan_id"], + model_id=raw["model_id"], + model_revision=raw["model_revision"], + prompts=tuple(PromptSpec(**prompt) for prompt in raw["prompts"]), + sampling=SamplingPolicy(**raw.get("sampling", {})), + concurrency_levels=tuple(raw.get("concurrency_levels", (1, 4))), + repeats=int(raw.get("repeats", 1)), + warmup_requests=int(raw.get("warmup_requests", 1)), + ) + + +def _recipe_from_config(spec: Mapping[str, Any]) -> RecipeSpec: + return RecipeSpec( + id=spec["id"], + runtime=spec["runtime"], + weight_format=spec["weight_format"], + weight_quantization=spec["weight_quantization"], + lane=Lane(spec["lane"]), + device=spec["device"], + artifact_path=spec.get("artifact_path", ""), + is_reference=bool(spec.get("is_reference", False)), + notes=spec.get("notes", ""), + ) + + +def run_configured_benchmark(config: Mapping[str, Any]) -> dict: + """Run every recipe in ``config`` against one shared plan and return the report. + + A recipe whose runtime cannot start is recorded as unavailable with the real + reason rather than dropped: a report that silently omits the recipe that + crashed would read as a clean result. + """ + require_real_inference() + plan = _plan_from_config(config) + + from .recipe_benchmark import RecipeMeasurement # local import keeps the seam obvious + + measurements = [] + for spec in config["recipes"]: + recipe = _recipe_from_config(spec) + try: + driver = build_driver(spec, plan) + measurements.append(measure_recipe(driver, recipe, plan)) + except Exception as exc: + measurements.append(RecipeMeasurement( + recipe=recipe, + load=LoadStats(artifact_bytes=0, load_ms=0.0), + unavailable_reason=f"{type(exc).__name__}: {exc}", + )) + + return build_report( + plan, + measurements, + host=dict(config.get("host", {})), + evidence_class=config.get("evidence_class", "local-real"), + ) diff --git a/tests/test_recipe_benchmark.py b/tests/test_recipe_benchmark.py new file mode 100644 index 0000000..3e3f370 --- /dev/null +++ b/tests/test_recipe_benchmark.py @@ -0,0 +1,310 @@ +"""The recipe benchmark's measurement core, driven by a scripted fake runtime. + +These tests never load a model, touch a GPU, or open a socket: the core is +deliberately runtime-free so the arithmetic and the lane rules can be pinned +down exactly, and the real drivers only have to be honest about what they +report. +""" + +from __future__ import annotations + +import pytest +from meshnet_node.recipe_benchmark import ( + BenchmarkError, + BenchmarkPlan, + GenerationSample, + Lane, + LoadStats, + PromptSpec, + RecipeSpec, + SamplingPolicy, + build_report, + compute_drift, + measure_recipe, + summarize_concurrency, + RequestOutcome, +) + +PROMPTS = ( + PromptSpec(id="short", text="Say hello.", context_class="short"), + PromptSpec(id="long", text="Summarize the following. " * 40, context_class="long"), +) + + +def plan(**overrides) -> BenchmarkPlan: + defaults = dict( + plan_id="test-plan", + model_id="test/model", + model_revision="revision-1", + prompts=PROMPTS, + sampling=SamplingPolicy(max_output_tokens=8), + concurrency_levels=(1, 4), + repeats=1, + warmup_requests=0, + ) + defaults.update(overrides) + return BenchmarkPlan(**defaults) + + +class FakeDriver: + """A runtime with fixed timings, so every metric below has one right answer.""" + + def __init__( + self, + *, + decode_ms_per_token: float = 10.0, + prefill_ms: float = 100.0, + artifact_bytes: int = 1_000_000, + rss_bytes: int = 4_000_000, + vram_bytes: int = 0, + texts: dict[str, str] | None = None, + fail_at_concurrency: int | None = None, + decode_tokens: int = 8, + ) -> None: + self.decode_ms_per_token = decode_ms_per_token + self.prefill_ms = prefill_ms + self.artifact_bytes = artifact_bytes + self.rss_bytes = rss_bytes + self.vram_bytes = vram_bytes + self.texts = texts or {} + self.fail_at_concurrency = fail_at_concurrency + self.decode_tokens = decode_tokens + self.in_flight = 0 + self.max_in_flight = 0 + self.loads = 0 + self.closes = 0 + self.generations = 0 + + def load(self) -> LoadStats: + self.loads += 1 + return LoadStats( + artifact_bytes=self.artifact_bytes, load_ms=50.0, + rss_bytes=self.rss_bytes, vram_bytes=self.vram_bytes, + ) + + def generate(self, prompt: str, sampling: SamplingPolicy) -> GenerationSample: + self.in_flight += 1 + self.max_in_flight = max(self.max_in_flight, self.in_flight) + try: + if self.fail_at_concurrency and self.in_flight >= self.fail_at_concurrency: + raise RuntimeError("slot exhausted") + self.generations += 1 + decode_ms = self.decode_ms_per_token * self.decode_tokens + return GenerationSample( + text=self.texts.get(prompt, "hello world"), + prompt_tokens=10, + decode_tokens=self.decode_tokens, + ttft_ms=self.prefill_ms, + prefill_ms=self.prefill_ms, + decode_ms=decode_ms, + total_ms=self.prefill_ms + decode_ms, + ) + finally: + self.in_flight -= 1 + + def memory_probe(self) -> tuple[int, int]: + return self.rss_bytes, self.vram_bytes + + def close(self) -> None: + self.closes += 1 + + +def recipe(recipe_id: str, lane: Lane, *, reference: bool = False, device: str = "cpu") -> RecipeSpec: + return RecipeSpec( + id=recipe_id, runtime="fake", weight_format="fake", weight_quantization="bf16", + lane=lane, device=device, is_reference=reference, + ) + + +def test_plan_rejects_an_experiment_it_cannot_run(): + with pytest.raises(BenchmarkError): + plan(prompts=()) + with pytest.raises(BenchmarkError): + plan(concurrency_levels=(0,)) + with pytest.raises(BenchmarkError): + plan(repeats=0) + + +def test_measure_runs_every_prompt_at_every_concurrency_level(): + driver = FakeDriver() + measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan()) + + # 2 prompts x (1 + 4) requests-per-level. + assert len(measurement.outcomes) == 2 * 1 + 2 * 4 + assert sorted(measurement.metrics) == [1, 4] + assert driver.loads == 1 + assert driver.closes == 1 + assert measurement.available + + +def test_concurrency_level_actually_overlaps_requests(): + driver = FakeDriver(decode_ms_per_token=5.0) + measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan(concurrency_levels=(4,))) + + assert driver.max_in_flight > 1, "concurrency 4 must run requests in parallel, not serially" + + +def test_driver_is_closed_even_when_every_request_fails(): + driver = FakeDriver(fail_at_concurrency=1) + measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan()) + + assert driver.closes == 1 + assert all(not outcome.ok for outcome in measurement.outcomes) + assert measurement.metrics[1].failures == 2 + assert measurement.metrics[1].failure_reasons == ("RuntimeError: slot exhausted",) + + +def test_failed_requests_are_reported_not_raised(): + driver = FakeDriver(fail_at_concurrency=4) + measurement = measure_recipe(driver, recipe("r", Lane.QUALITY, reference=True), plan()) + + assert measurement.metrics[1].failures == 0 + assert measurement.metrics[4].failures > 0 + assert measurement.metrics[4].requests == 8 + + +def test_summary_arithmetic_is_exact(): + outcomes = [ + RequestOutcome( + recipe_id="r", concurrency=2, prompt_id="p", repeat=0, ok=True, + latency_ms=200.0, ttft_ms=100.0, prefill_ms=100.0, decode_ms=100.0, + prompt_tokens=10, decode_tokens=10, + ), + RequestOutcome( + recipe_id="r", concurrency=2, prompt_id="p", repeat=1, ok=True, + latency_ms=400.0, ttft_ms=200.0, prefill_ms=200.0, decode_ms=200.0, + prompt_tokens=10, decode_tokens=10, + ), + ] + metrics = summarize_concurrency( + outcomes, concurrency=2, wall_ms=1000.0, peak_rss_bytes=7, peak_vram_bytes=9 + ) + + assert metrics.latency_p50_ms == 200.0 + assert metrics.latency_p95_ms == 400.0 + # 10 tok / 0.1 s = 100 tok/s and 10 tok / 0.2 s = 50 tok/s, averaged. + assert metrics.decode_tokens_per_sec == 75.0 + # 20 decoded tokens over a 1 s wall clock, regardless of per-request rates. + assert metrics.aggregate_decode_tokens_per_sec == 20.0 + assert (metrics.peak_rss_bytes, metrics.peak_vram_bytes) == (7, 9) + + +def test_aggregate_throughput_credits_overlap_but_per_request_rate_does_not(): + """Two runtimes with identical per-request speed must be told apart by overlap.""" + serial = summarize_concurrency( + [ + RequestOutcome(recipe_id="s", concurrency=4, prompt_id="p", repeat=i, ok=True, + latency_ms=100.0, decode_ms=100.0, decode_tokens=10) + for i in range(4) + ], + concurrency=4, wall_ms=400.0, peak_rss_bytes=0, peak_vram_bytes=0, + ) + batched = summarize_concurrency( + [ + RequestOutcome(recipe_id="b", concurrency=4, prompt_id="p", repeat=i, ok=True, + latency_ms=100.0, decode_ms=100.0, decode_tokens=10) + for i in range(4) + ], + concurrency=4, wall_ms=100.0, peak_rss_bytes=0, peak_vram_bytes=0, + ) + + assert serial.decode_tokens_per_sec == batched.decode_tokens_per_sec == 100.0 + assert serial.aggregate_decode_tokens_per_sec == 100.0 + assert batched.aggregate_decode_tokens_per_sec == 400.0 + + +def test_drift_against_the_reference_is_exact_for_an_identical_runtime(): + texts = {prompt.text: f"answer for {prompt.id}" for prompt in PROMPTS} + reference = measure_recipe( + FakeDriver(texts=texts), recipe("ref", Lane.QUALITY, reference=True), plan() + ) + twin = measure_recipe(FakeDriver(texts=texts), recipe("twin", Lane.QUALITY), plan()) + + drift = compute_drift(twin, reference) + assert drift.compared_prompts == 2 + assert drift.exact_match_rate == 1.0 + assert drift.mean_similarity == 1.0 + assert drift.advisory is False + + +def test_quantized_drift_is_advisory_and_never_an_equivalence_claim(): + reference = measure_recipe( + FakeDriver(texts={prompt.text: "the capital is Paris" for prompt in PROMPTS}), + recipe("ref", Lane.QUALITY, reference=True), plan(), + ) + quantized = measure_recipe( + FakeDriver(texts={prompt.text: "the capital is Lyon" for prompt in PROMPTS}), + recipe("q4", Lane.PERFORMANCE_FIT), plan(), + ) + + drift = compute_drift(quantized, reference) + assert drift.advisory is True, "a quantized recipe's drift must be advisory" + assert drift.exact_match_rate == 0.0 + assert 0.0 < drift.mean_similarity < 1.0 + assert drift.per_prompt[0]["first_divergence_char"] > 0 + + +def test_report_needs_exactly_one_quality_lane_reference(): + measurement = measure_recipe(FakeDriver(), recipe("a", Lane.QUALITY, reference=True), plan()) + second = measure_recipe(FakeDriver(), recipe("b", Lane.QUALITY, reference=True), plan()) + quantized = measure_recipe(FakeDriver(), recipe("q", Lane.PERFORMANCE_FIT), plan()) + + with pytest.raises(BenchmarkError, match="exactly one reference"): + build_report(plan(), [measurement, second], host={}, evidence_class="synthetic") + with pytest.raises(BenchmarkError, match="exactly one reference"): + build_report(plan(), [quantized], host={}, evidence_class="synthetic") + + +def test_reference_recipe_may_not_be_quantized(): + quantized_reference = measure_recipe( + FakeDriver(), recipe("q", Lane.PERFORMANCE_FIT, reference=True), plan() + ) + with pytest.raises(BenchmarkError, match="quality lane"): + build_report(plan(), [quantized_reference], host={}, evidence_class="synthetic") + + +def test_report_must_declare_how_it_was_produced(): + measurement = measure_recipe(FakeDriver(), recipe("a", Lane.QUALITY, reference=True), plan()) + with pytest.raises(BenchmarkError, match="evidence class"): + build_report(plan(), [measurement], host={}, evidence_class="probably-real") + + +def test_report_carries_every_metric_the_contract_reads(): + reference = measure_recipe(FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan()) + quantized = measure_recipe( + FakeDriver(decode_ms_per_token=4.0, artifact_bytes=400_000, rss_bytes=1_000_000), + recipe("q4", Lane.PERFORMANCE_FIT), plan(), + ) + report = build_report( + plan(), [reference, quantized], host={"cpu": "test"}, evidence_class="synthetic" + ) + + assert report["schema_version"] == 1 + assert report["reference_recipe_id"] == "ref" + entry = next(e for e in report["recipes"] if e["recipe"]["id"] == "q4") + cell = entry["concurrency"]["1"] + for metric in ( + "ttft_p50_ms", "ttft_p95_ms", "latency_p50_ms", "latency_p95_ms", + "prefill_tokens_per_sec", "decode_tokens_per_sec", "aggregate_decode_tokens_per_sec", + "peak_rss_bytes", "peak_vram_bytes", "failures", + ): + assert metric in cell, f"the contract reads {metric}, so the report must carry it" + assert entry["load"]["artifact_bytes"] == 400_000 + assert [d["recipe_id"] for d in report["drift"]] == ["q4"] + + +def test_unavailable_recipes_are_recorded_rather_than_dropped(): + from meshnet_node.recipe_benchmark import RecipeMeasurement + + reference = measure_recipe(FakeDriver(), recipe("ref", Lane.QUALITY, reference=True), plan()) + missing = RecipeMeasurement( + recipe=recipe("q4", Lane.PERFORMANCE_FIT), + load=LoadStats(artifact_bytes=0, load_ms=0.0), + unavailable_reason="BenchmarkError: GGUF artifact not found", + ) + report = build_report(plan(), [reference, missing], host={}, evidence_class="synthetic") + + entry = next(e for e in report["recipes"] if e["recipe"]["id"] == "q4") + assert entry["available"] is False + assert "not found" in entry["unavailable_reason"] + assert report["drift"] == [], "an unmeasured recipe has no drift to report"