From a508768e8ad79051b2b1d6529cc1bc45841b98fc Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 14 Jul 2026 22:46:11 +0300 Subject: [PATCH] feat: add live endpoint benchmark runner --- .../evidence/DGR-001/README.md | 5 +- .../node/meshnet_node/performance_contract.py | 97 +++++++++++++++++++ tests/test_performance_contract.py | 36 +++++++ 3 files changed, 136 insertions(+), 2 deletions(-) diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-001/README.md b/.scratch/distributed-gguf-runtime/evidence/DGR-001/README.md index 33e8121..8f0f5ea 100644 --- a/.scratch/distributed-gguf-runtime/evidence/DGR-001/README.md +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-001/README.md @@ -63,8 +63,9 @@ Result: passed ## Limitations -- This slice captures the DGR-001 contract and baseline selection only. -- It does **not** download or run a real model yet. +- This slice still uses a deterministic stub backend for the core comparison matrix. +- It now also includes a live endpoint runner that can fan out one OpenAI-compatible request per lane when the caller provides endpoints. +- It does **not** download or run a real model from within the repo. - Real safetensors vs GGUF execution, TTFT/prefill/decode measurements, RSS/VRAM capture, and output-drift comparison are still to be implemented against the contract. ## Compatibility notes diff --git a/packages/node/meshnet_node/performance_contract.py b/packages/node/meshnet_node/performance_contract.py index 4c035d8..724765f 100644 --- a/packages/node/meshnet_node/performance_contract.py +++ b/packages/node/meshnet_node/performance_contract.py @@ -12,8 +12,11 @@ from __future__ import annotations import argparse import json +import time +import urllib.request from dataclasses import dataclass from pathlib import Path +from typing import Mapping SCHEMA_VERSION = 1 CONTRACT_ID = "DGR-001" @@ -330,6 +333,100 @@ def run_performance_benchmark( } +def run_real_model_endpoint_benchmark( + endpoints: Mapping[str, str], + *, + model: str, + contract: PerformanceContract = DEFAULT_CONTRACT, + timeout: float = 120.0, +) -> dict: + """Run one live OpenAI-compatible request per lane against supplied endpoints. + + The caller provides one URL per benchmark lane. The runner measures the + request/response round-trip at the client boundary and reuses the same + contract schema as the deterministic stub. + """ + + def _sample_for_lane(lane: BenchmarkLane, endpoint: str) -> LaneSample: + prompt = " ".join(contract.model_target.rationale.split()[:6]) + body = json.dumps( + { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "max_tokens": len(STUB_OUTPUT_TOKENS), + "temperature": 0, + } + ).encode("utf-8") + request = urllib.request.Request( + f"{endpoint.rstrip('/')}/v1/chat/completions", + data=body, + headers={ + "Content-Type": "application/json", + "X-Meshnet-Lane": lane.id, + }, + method="POST", + ) + started = time.monotonic() + with urllib.request.urlopen(request, timeout=timeout) as response: + response_body = response.read() + session_id = response.headers.get("X-Meshnet-Session", f"{lane.id}-session") + elapsed_ms = round((time.monotonic() - started) * 1000, 4) + payload = json.loads(response_body) + content = payload["choices"][0]["message"]["content"] + tokens = tuple(content.split()) + token_count = max(1, len(tokens)) + artifact_gb = ( + contract.model_target.gguf_size_gb + if lane.runtime == "llama.cpp" + else _SAFETENSORS_BF16_ARTIFACT_GB + ) + return LaneSample( + ttft_ms=elapsed_ms, + prefill_tok_per_sec=round(token_count / max(0.001, elapsed_ms / 1000), 4), + decode_tok_per_sec=round(token_count / max(0.001, elapsed_ms / 1000), 4), + rss_bytes=0, + vram_bytes=0, + artifact_bytes=_gb(artifact_gb), + output_tokens=tokens, + ) + + lanes = [] + for lane in contract.benchmark_lanes: + if lane.id not in endpoints: + raise KeyError(f"missing endpoint for lane {lane.id}") + lanes.append((lane, _sample_for_lane(lane, endpoints[lane.id]))) + references = { + lane.device: sample.output_tokens for lane, sample in lanes if lane.runtime == "transformers" + } + lane_reports = [] + for lane, sample in lanes: + drift = _output_drift(sample.output_tokens, references.get(lane.device, sample.output_tokens)) + lane_reports.append({ + **lane.to_dict(), + "output_tokens": list(sample.output_tokens), + "results": [ + {"concurrency": level, "metrics": _metrics_for(sample, level, drift)} + for level in lane.concurrency_levels + ], + }) + devices = sorted({lane.device for lane, _ in lanes}) + comparisons = {device: _compare_device(lanes, device) for device in devices} + gguf_benefit = any(comparison["gguf_benefit"] for comparison in comparisons.values()) + return { + "schema_version": BENCHMARK_SCHEMA_VERSION, + "story_id": contract.story_id, + "source": "real-model-endpoints", + "model_target": contract.model_target.to_dict(), + "lanes": lane_reports, + "comparisons": comparisons, + "stop_condition": { + "text": contract.stop_condition, + "gguf_benefit": gguf_benefit, + "triggered": not gguf_benefit, + }, + } + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Write the DGR-001 performance contract JSON") parser.add_argument("--json-out", type=Path, default=DEFAULT_OUTPUT_PATH, help="output JSON path") diff --git a/tests/test_performance_contract.py b/tests/test_performance_contract.py index 8267227..537d2e3 100644 --- a/tests/test_performance_contract.py +++ b/tests/test_performance_contract.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +from unittest.mock import MagicMock, patch from meshnet_node.performance_contract import ( BENCHMARK_SCHEMA_VERSION, @@ -10,6 +11,7 @@ from meshnet_node.performance_contract import ( SCHEMA_VERSION, main, run_performance_benchmark, + run_real_model_endpoint_benchmark, ) @@ -165,3 +167,37 @@ def test_contract_cli_writes_benchmark_report(tmp_path, capsys): output = capsys.readouterr().out assert str(contract_out) in output assert str(benchmark_out) in output + + +def test_real_model_endpoint_benchmark_uses_lane_specific_endpoints_and_shared_schema(): + """The live client path fans out to one endpoint per CPU/GPU lane. + + Tags: performance, benchmark, live + """ + response = MagicMock() + response.read.return_value = json.dumps({"choices": [{"message": {"content": "mesh activation"}}]}).encode() + response.headers.get.return_value = "lane-session" + response.__enter__.return_value = response + + endpoints = { + "transformers-safetensors-cpu": "http://cpu-safetensors", + "llama-cpp-gguf-cpu": "http://cpu-gguf", + "transformers-safetensors-gpu": "http://gpu-safetensors", + "llama-cpp-gguf-gpu": "http://gpu-gguf", + } + + with patch("meshnet_node.performance_contract.urllib.request.urlopen", return_value=response) as urlopen: + report = run_real_model_endpoint_benchmark(endpoints=endpoints, model="deepseek-ai/DeepSeek-V2-Lite-Chat") + + assert report["source"] == "real-model-endpoints" + assert report["model_target"] == DEFAULT_CONTRACT.model_target.to_dict() + assert set(report["comparisons"]) == {"cpu", "gpu"} + assert urlopen.call_count == len(endpoints) + called_urls = [call.args[0].full_url for call in urlopen.call_args_list] + assert called_urls == [f"{url}/v1/chat/completions" for url in endpoints.values()] + for lane in report["lanes"]: + assert lane["results"][0]["metrics"]["decode_tok_per_sec"] > 0 + assert lane["results"][0]["metrics"]["ttft_ms"] > 0 + assert lane["output_tokens"] == ["mesh", "activation"] + assert report["comparisons"]["cpu"]["gguf_lane"] == "llama-cpp-gguf-cpu" + assert report["comparisons"]["gpu"]["gguf_lane"] == "llama-cpp-gguf-gpu"