From a78b175468a082faf73da24a271e043695d61ccb Mon Sep 17 00:00:00 2001 From: Helder Vasconcelos Date: Thu, 11 Jun 2026 19:52:34 +0100 Subject: [PATCH 1/3] server: terminate SSE streams with sink.done() instead of aborting Both streaming endpoints (OpenAI chat and Anthropic messages) ended their chunked content provider by returning false after the final frame. In cpp-httplib that aborts the connection without writing the chunked-encoding terminator, so strict HTTP/1.1 clients (httpx/h11, hence the OpenAI Python SDK) reported a protocol error at the end of every streamed response even after seeing [DONE]. Call sink.done() and return true; mid-stream client-disconnect aborts keep returning false. Found by the new bench/ harness on its first run. Co-Authored-By: Claude Fable 5 --- src/server/http_server.cpp | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/server/http_server.cpp b/src/server/http_server.cpp index 04bcb68..0a98ced 100644 --- a/src/server/http_server.cpp +++ b/src/server/http_server.cpp @@ -215,10 +215,15 @@ void HttpServer::stream_chat(const std::shared_ptr& req, httplib::Respo return false; // trailing text from detok.finish() } - // Final chunk carries the finish_reason, then the [DONE] sentinel. + // Final chunk carries the finish_reason, then the [DONE] sentinel. The + // stream must end via sink.done() (writes the chunked-encoding terminator); + // returning false instead aborts the connection mid-body, which strict + // HTTP/1.1 clients (httpx/h11, hence the OpenAI SDK) reject as a protocol + // error even after seeing [DONE]. send(sse_frame(make_chat_chunk(id, created, model, json::object(), finish))); send(kSseDone); - return false; // stream complete + sink.done(); + return true; }); } @@ -364,7 +369,8 @@ void HttpServer::stream_messages(const std::shared_ptr& req, httplib::R if (!send(sse_event("message_delta", make_message_delta(stop_reason, output_tokens)))) return false; send(sse_event("message_stop", kMessageStop)); - return false; // stream complete + sink.done(); // terminate the chunked body cleanly (see stream_chat) + return true; }); } From fbacdf85b70a0d8f70bac88fc2245204753f8555 Mon Sep 17 00:00:00 2001 From: Helder Vasconcelos Date: Thu, 11 Jun 2026 19:52:51 +0100 Subject: [PATCH 2/3] Add cross-engine benchmark harness (bench/) A Python orchestrator that measures the engine under different configurations and compares it against the other local engines on Apple Silicon (llama.cpp's llama-server, vllm-mlx, omlx), all driven identically through their OpenAI-compatible streaming APIs, one engine at a time. - Scenarios: concurrency sweep (1/4/8/16 streams), prompt-length sweep (128-8k tokens), mlxforge-only kv_bits x prefix_cache sweep with /health counter deltas, and cross-engine multi-turn shared-prefix reuse. - Engines auto-launch with matched settings and are skipped with a report note when not installed; launch argv is TOML-overridable. llama-server gets per-scenario right-sized -np/-c (its KV buffer is preallocated). - Methodology: deterministic synthesized prompts with per-engine token calibration (no tokenizer dep), unique per-request prefixes so prompt caches can't fake prefill numbers (multi-turn shares one on purpose), discarded warmups per launch, temperature 0 + ignore_eos on llama.cpp, dual token accounting (server usage vs client chunk count, flagged per row). - Output: self-contained HTML report (best value per comparable group highlighted) + raw JSON; --rerender regenerates the HTML; plain-text tables on stdout. Runs via `uv run bench/bench.py` (PEP 723, httpx only); --quick gives a minutes-long smoke run. Like the server and CLI, this is a QA harness, not a product deliverable. Co-Authored-By: Claude Fable 5 --- .gitignore | 4 + bench/README.md | 91 ++++++++++++++ bench/bench.py | 214 ++++++++++++++++++++++++++++++++ bench/config.py | 127 +++++++++++++++++++ bench/engines.py | 288 ++++++++++++++++++++++++++++++++++++++++++++ bench/loadgen.py | 227 ++++++++++++++++++++++++++++++++++ bench/prompts.py | 73 +++++++++++ bench/report.py | 262 ++++++++++++++++++++++++++++++++++++++++ bench/scenarios.py | 261 +++++++++++++++++++++++++++++++++++++++ doc/applications.md | 17 +++ 10 files changed, 1564 insertions(+) create mode 100644 bench/README.md create mode 100644 bench/bench.py create mode 100644 bench/config.py create mode 100644 bench/engines.py create mode 100644 bench/loadgen.py create mode 100644 bench/prompts.py create mode 100644 bench/report.py create mode 100644 bench/scenarios.py diff --git a/.gitignore b/.gitignore index 35a1cbf..4b99e4b 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,10 @@ __pycache__/ *.pyc +# Benchmark harness output (the harness itself is committed, results are not) +/bench/results/ +/bench/.venv/ + # Editor / OS .DS_Store .idea/ diff --git a/bench/README.md b/bench/README.md new file mode 100644 index 0000000..5e6117d --- /dev/null +++ b/bench/README.md @@ -0,0 +1,91 @@ +# bench — cross-engine benchmark harness + +Measures mlxforge against the other local-inference engines on Apple Silicon +(**llama.cpp**'s `llama-server`, **vllm-mlx**, **omlx**) and sweeps mlxforge's +own configuration space (KV quantization, prefix cache). Every engine is +driven the same way — through its OpenAI-compatible `/v1/chat/completions` +SSE endpoint — so the comparison is apples-to-apples, and engines run +strictly one at a time (they all contend for the same Metal GPU). + +This is a QA harness, like the server and CLI: it exists to characterize and +prove the engine, not as a product deliverable. + +## Run + +```sh +uv run bench/bench.py --quick --engines mlxforge --models qwen3-0.6b # ~3 min smoke +uv run bench/bench.py # full run +uv run bench/bench.py --rerender bench/results/bench-.json # re-render report +``` + +No `uv`? `python3 -m venv bench/.venv && bench/.venv/bin/pip install httpx` +then `bench/.venv/bin/python bench/bench.py ...` (needs Python ≥ 3.11). + +`build/mlxforge` must exist (`cmake --build build --parallel`). Engines that +aren't installed are skipped with a note in the report: + +- llama.cpp: `brew install llama.cpp` (provides `llama-server`) +- vllm-mlx: `pip install vllm-mlx` +- omlx: see its install docs; models must be present in its model dir + (set `omlx_model_dir` in a TOML config if not using omlx's default) + +Models download from HuggingFace on first use (each engine into its own +cache), so the first run per engine×model is slow — `--ready-timeout 900` +helps on slow links. Pre-pulling is recommended for clean timing. + +## Scenarios + +| name | what it measures | +|---|---| +| `concurrency` | aggregate tok/s + per-request TTFT/decode at 1/4/8/16 concurrent streams — the continuous-batching story | +| `prompt` | single-stream TTFT and prefill tok/s at 128/512/2k/8k prompt tokens | +| `config` | mlxforge only: kv_bits {0,8,4} × prefix_cache {off,on}, each a server relaunch, with `/health` counter deltas | +| `multiturn` | cross-engine: chat turns sharing a ~2k-token system prompt; cold vs warm TTFT (prompt/prefix caches) | + +Prompts are deterministic synthesized word streams; target token counts are +hit by calibrating tokens-per-word against each server's reported +`usage.prompt_tokens` (no Python tokenizer dependency). Scenarios 1–2 give +every request a unique lead-in so prompt caches can't fake prefill numbers; +`multiturn` shares a prefix on purpose. Requests use `temperature 0`, fixed +seed, and an open-ended task (plus `ignore_eos` on llama.cpp) so decode +lengths are comparable; throughput always uses *actual* completion tokens. + +## Output + +Plain-text tables on stdout, a self-contained HTML report +(`bench/results/bench-.html`, no external assets — best value per +comparable group highlighted), and the raw results as JSON next to it; +`--rerender FILE.json` regenerates the HTML next to the JSON. Token counts are flagged per row: `server` = +engine-reported usage, `client` = SSE content-chunk count (mlxforge streams +no usage chunk, so its rows are client-counted; aggregates can be +cross-checked against `/health`). + +## Configuration + +Defaults are embedded in `config.py`; `--config my.toml` overrides them: + +```toml +ready_timeout = 900 +omlx_model_dir = "/Users/me/.omlx/models" + +[models.llama1b-4bit] +gguf_path = "/path/to/Llama-3.2-1B-Instruct-Q4_K_M.gguf" # skip -hf download + +# If an external engine's CLI drifts from the assumptions baked into +# engines.py, override the full argv ({model}/{port}/... placeholders): +[engines.vllm-mlx] +cmd = ["vllm-mlx", "serve", "{model}", "--port", "{port}"] +``` + +Assumed external launch commands (override as above if they drift): +`llama-server -hf -c -np -ngl 99 --no-webui`, +`vllm-mlx serve --continuous-batching`, `omlx serve [--model-dir D]`. + +## Caveats + +- GGUF quants (Q4_K_M) are the *nearest* equivalents to MLX 4-bit group + quants, not identical — cross-engine 4-bit rows are approximate by nature. +- Two discarded warmup generations follow every server launch (Metal JIT); + the first concurrency level also runs a discarded extra pass. +- One engine at a time, always; never compare numbers from a run where + anything else was using the GPU. diff --git a/bench/bench.py b/bench/bench.py new file mode 100644 index 0000000..8f40fff --- /dev/null +++ b/bench/bench.py @@ -0,0 +1,214 @@ +# /// script +# requires-python = ">=3.11" +# dependencies = ["httpx"] +# /// +"""Cross-engine inference benchmark harness. + +Drives mlxforge, llama.cpp (llama-server), vllm-mlx, and omlx through their +OpenAI-compatible HTTP APIs — strictly one engine at a time (they all contend +for the same Metal GPU) — and renders a text report plus raw JSON results. + + uv run bench/bench.py --quick --engines mlxforge --models qwen3-0.6b + uv run bench/bench.py # full run, all detected engines + uv run bench/bench.py --rerender results/bench-.json + +See bench/README.md for engine install hints and launch-command assumptions. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import shutil +import sys +from datetime import datetime +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import report +import scenarios as sc +from config import ENGINE_NAMES, SCENARIO_NAMES, BenchConfig, ModelSpec, load_config +from engines import EngineAdapter, ServerOpts, make_adapters +from loadgen import make_client + +# Scenarios each engine can run; "config" is the mlxforge-internal sweep. +CROSS_ENGINE_SCENARIOS = ["concurrency", "prompt", "multiturn"] + + +def parse_args() -> argparse.Namespace: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--engines", default=",".join(ENGINE_NAMES), + help=f"comma list of {ENGINE_NAMES} (missing ones auto-skip)") + ap.add_argument("--scenarios", default=",".join(SCENARIO_NAMES), + help=f"comma list of {SCENARIO_NAMES}") + ap.add_argument("--models", default="", + help="comma list of model keys (default: all configured)") + ap.add_argument("--config", type=Path, default=None, help="optional TOML overrides") + ap.add_argument("--out-dir", type=Path, default=None) + ap.add_argument("--quick", action="store_true", help="tiny counts; smoke run in minutes") + ap.add_argument("--mlxforge-bin", type=Path, default=None) + ap.add_argument("--ready-timeout", type=float, default=None, + help="server readiness timeout in seconds (first run may download models)") + ap.add_argument("--rerender", type=Path, default=None, + help="re-render the text report from a saved results JSON and exit") + ap.add_argument("--keep-logs", action="store_true", + help="keep per-engine server logs even when every run succeeds") + return ap.parse_args() + + +def plan_launches(engine: str, model: ModelSpec, scenario_names: list[str], + cfg: BenchConfig, config_model: str) -> list[tuple[ServerOpts, list[str]]]: + """Map (engine, model, scenarios) to server launches: most engines need one + launch for everything; mlxforge relaunches per config (prefix cache and the + kv_bits sweep are engine-creation settings, not request settings).""" + max_conc = max(cfg.params.concurrency_levels + [cfg.params.config_probe_concurrency]) + cross = [s for s in scenario_names if s in CROSS_ENGINE_SCENARIOS] + if engine == "llamacpp": + # llama-server preallocates -c × -np of KV, so each launch is sized to + # its scenario: many small slots for the concurrency sweep, one + # full-context slot for the prompt sweep and multi-turn. + launches = [] + if "concurrency" in cross: + need = cfg.params.concurrency_prompt_tokens + cfg.params.concurrency_max_tokens + 384 + slot = -(-need // 1024) * 1024 # round up to 1k + launches.append((ServerOpts(label="batch", max_concurrency=max_conc, + ctx_per_slot=slot), ["concurrency"])) + rest = [s for s in cross if s != "concurrency"] + if rest: + launches.append((ServerOpts(max_concurrency=1), rest)) + return launches + if engine != "mlxforge": + return [(ServerOpts(max_concurrency=max_conc), cross)] if cross else [] + + launches: list[tuple[ServerOpts, list[str]]] = [] + plain = [s for s in cross if s != "multiturn"] + if plain: + launches.append((ServerOpts(max_concurrency=max_conc), plain)) + if "multiturn" in cross: + # The shared-prefix scenario is what the prefix cache exists for. + launches.append((ServerOpts(label="prefix", prefix_cache=True, + max_concurrency=max_conc), ["multiturn"])) + if "config" in scenario_names and model.key == config_model: + for bits in cfg.params.config_kv_bits: + for prefix in (False, True): + launches.append((ServerOpts(label=f"kv{bits}-{'pfx' if prefix else 'nopfx'}", + kv_bits=bits, prefix_cache=prefix, + max_concurrency=max_conc), ["config"])) + return launches + + +async def run_launch(adapter: EngineAdapter, model: ModelSpec, opts: ServerOpts, + scenario_names: list[str], cfg: BenchConfig, quick: bool, + log_dir: Path) -> list[dict]: + runs: list[dict] = [] + ep = adapter.launch(model, opts, log_dir) + try: + adapter.wait_ready(ep, model, cfg.ready_timeout) + async with make_client() as client: + ctx = sc.ServerCtx( + adapter=adapter, ep=ep, model_name=adapter.api_model_name(model), + opts=opts, client=client, extra=adapter.extra_request_fields(opts)) + ctx.cal = await sc.calibrate(ctx) + print(f" calibrated: {ctx.cal.tokens_per_word:.2f} tok/word, " + f"overhead {ctx.cal.overhead_tokens:.0f} tok", flush=True) + await sc.warmup(ctx) + for name in scenario_names: + print(f" scenario: {name}", flush=True) + rows = await sc.RUNNERS[name](ctx, cfg.params, quick) + runs.append({"scenario": name, "engine": adapter.name, "model": model.key, + "server_config": opts.as_dict(), "params": vars(cfg.params), + "rows": rows}) + except Exception as e: + note = f"{type(e).__name__}: {e}" + print(f" FAILED: {note}", file=sys.stderr, flush=True) + runs.append({"scenario": "/".join(scenario_names), "engine": adapter.name, + "model": model.key, "server_config": opts.as_dict(), + "rows": [], "note": note}) + finally: + adapter.stop(ep) + return runs + + +async def main() -> int: + args = parse_args() + if args.rerender: + results = json.loads(args.rerender.read_text()) + html_path = args.rerender.with_suffix(".html") + html_path.write_text(report.render_html(results)) + print(report.render(results), end="") + print(f"\nreport: {html_path}") + return 0 + + cfg = load_config(args.config) + if args.mlxforge_bin: + cfg.mlxforge_bin = args.mlxforge_bin + if args.ready_timeout is not None: + cfg.ready_timeout = args.ready_timeout + if args.out_dir: + cfg.out_dir = args.out_dir + if args.models: + keys = args.models.split(",") + unknown = [m for m in keys if m not in cfg.models] + if unknown: + print(f"unknown model keys: {unknown} (known: {list(cfg.models)})", file=sys.stderr) + return 2 + cfg.model_keys = keys + if args.quick: + cfg.params = cfg.params.quick() + + engine_names = [e.strip() for e in args.engines.split(",") if e.strip()] + scenario_names = [s.strip() for s in args.scenarios.split(",") if s.strip()] + for name, known in ((engine_names, ENGINE_NAMES), (scenario_names, SCENARIO_NAMES)): + bad = [n for n in name if n not in known] + if bad: + print(f"unknown name(s): {bad} (known: {known})", file=sys.stderr) + return 2 + + models = cfg.selected_models() + config_model = (cfg.params.config_sweep_model + if any(m.key == cfg.params.config_sweep_model for m in models) + else models[0].key) + + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + log_dir = cfg.out_dir / f"logs-{ts}" + adapters = make_adapters(cfg) + engines_info: dict[str, dict] = {} + runs: list[dict] = [] + for name in engine_names: + adapter = adapters[name] + version = adapter.detect() + engines_info[name] = {"detected": version is not None, "version": version, + "note": adapter.skip_note or None} + if version is None: + print(f"== {name}: SKIPPED ({adapter.skip_note})", flush=True) + continue + print(f"== {name} ({version})", flush=True) + for model in models: + for opts, names in plan_launches(name, model, scenario_names, cfg, config_model): + print(f" {model.key} [{opts.label}] -> {','.join(names)}", flush=True) + runs += await run_launch(adapter, model, opts, names, cfg, args.quick, log_dir) + + results = { + "meta": {"timestamp": datetime.now().isoformat(timespec="seconds"), + "host": report.host_info(), "quick": args.quick, + "argv": sys.argv[1:]}, + "engines": engines_info, + "runs": runs, + } + html_path, json_path = report.save(results, cfg.out_dir) + print() + print(report.render(results), end="") + print(f"\nreport: {html_path}\nraw: {json_path}") + + had_failures = any(r.get("note") for r in runs) + if log_dir.exists() and not (args.keep_logs or had_failures): + shutil.rmtree(log_dir) + elif log_dir.exists(): + print(f"logs: {log_dir}") + return 1 if had_failures else 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/bench/config.py b/bench/config.py new file mode 100644 index 0000000..4db699b --- /dev/null +++ b/bench/config.py @@ -0,0 +1,127 @@ +"""Benchmark configuration: embedded defaults + optional TOML overrides. + +The full default config lives here as dataclasses so the harness runs with no +config file at all; `--config FILE.toml` merges overrides on top (stdlib +tomllib, so TOML can carry comments without adding a dependency). +""" + +from __future__ import annotations + +import tomllib +from dataclasses import dataclass, field, replace +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent + +ENGINE_NAMES = ["mlxforge", "llamacpp", "vllm-mlx", "omlx"] +SCENARIO_NAMES = ["concurrency", "prompt", "config", "multiturn"] + + +@dataclass(frozen=True) +class ModelSpec: + key: str + mlx: str # HF repo id for the MLX engines (mlxforge, vllm-mlx, omlx) + gguf: str # llama-server -hf spec ("repo:quant"); approximate quant match + gguf_path: str = "" # local .gguf overrides -hf when set + omlx_name: str = "" # model name omlx advertises; default = mlx repo basename + + @property + def omlx_model(self) -> str: + return self.omlx_name or self.mlx.rsplit("/", 1)[-1] + + +@dataclass +class ScenarioParams: + concurrency_levels: list[int] = field(default_factory=lambda: [1, 4, 8, 16]) + concurrency_prompt_tokens: int = 512 + concurrency_max_tokens: int = 128 + prompt_sizes: list[int] = field(default_factory=lambda: [128, 512, 2048, 8192]) + prompt_repeats: int = 3 + prompt_max_tokens: int = 32 + config_sweep_model: str = "llama1b-4bit" + config_kv_bits: list[int] = field(default_factory=lambda: [0, 8, 4]) + config_probe_concurrency: int = 8 + multiturn_system_tokens: int = 2048 + multiturn_turns: int = 5 # 1 cold + (turns-1) warm + multiturn_max_tokens: int = 64 + multiturn_passes: int = 2 # fresh system prompt per pass to validate cold numbers + + def quick(self) -> "ScenarioParams": + # Tiny counts for a smoke run: minutes, not an evaluation. + return replace( + self, + concurrency_levels=[1, 4], + concurrency_max_tokens=32, + prompt_sizes=[128, 512], + prompt_repeats=1, + config_kv_bits=[0], # 2 relaunches (prefix on/off) instead of 6 + config_probe_concurrency=4, + multiturn_system_tokens=512, + multiturn_turns=3, + multiturn_max_tokens=16, + multiturn_passes=1, + ) + + +@dataclass +class BenchConfig: + models: dict[str, ModelSpec] = field(default_factory=dict) + engines: list[str] = field(default_factory=lambda: list(ENGINE_NAMES)) + scenarios: list[str] = field(default_factory=lambda: list(SCENARIO_NAMES)) + model_keys: list[str] = field(default_factory=list) # empty = all in `models` + params: ScenarioParams = field(default_factory=ScenarioParams) + out_dir: Path = REPO_ROOT / "bench" / "results" + mlxforge_bin: Path = REPO_ROOT / "build" / "mlxforge" + ready_timeout: float = 300.0 # first launch may include an HF download + max_ctx: int = 16384 + omlx_model_dir: str = "" # default: omlx's own default (~/.omlx/models) + # Per-engine argv template overrides (TOML [engines.] cmd = [...]), + # formatted with {model} {port} {ctx} {np} {kv_bits} {prefix_cache} {model_dir}. + engine_cmds: dict[str, list[str]] = field(default_factory=dict) + + def selected_models(self) -> list[ModelSpec]: + keys = self.model_keys or list(self.models) + return [self.models[k] for k in keys] + + +DEFAULT_MODELS = { + "llama1b-bf16": ModelSpec( + key="llama1b-bf16", + mlx="mlx-community/Llama-3.2-1B-Instruct-bf16", + gguf="bartowski/Llama-3.2-1B-Instruct-GGUF:f16", + ), + "llama1b-4bit": ModelSpec( + key="llama1b-4bit", + mlx="mlx-community/Llama-3.2-1B-Instruct-4bit", + # Nearest available quant — MLX 4-bit group quant != Q4_K_M (noted in report). + gguf="bartowski/Llama-3.2-1B-Instruct-GGUF:Q4_K_M", + ), + "qwen3-0.6b": ModelSpec( + key="qwen3-0.6b", + mlx="mlx-community/Qwen3-0.6B-bf16", + gguf="ggml-org/Qwen3-0.6B-GGUF:f16", + ), +} + + +def load_config(toml_path: Path | None = None) -> BenchConfig: + cfg = BenchConfig(models=dict(DEFAULT_MODELS)) + if toml_path is None: + return cfg + data = tomllib.loads(Path(toml_path).read_text()) + for key, m in data.get("models", {}).items(): + base = cfg.models.get(key, ModelSpec(key=key, mlx="", gguf="")) + cfg.models[key] = replace(base, **{f: m[f] for f in ("mlx", "gguf", "gguf_path", "omlx_name") if f in m}) + for name, e in data.get("engines", {}).items(): + if "cmd" in e: + cfg.engine_cmds[name] = list(e["cmd"]) + for f in ("ready_timeout", "max_ctx", "omlx_model_dir"): + if f in data: + setattr(cfg, f, data[f]) + if "mlxforge_bin" in data: + cfg.mlxforge_bin = Path(data["mlxforge_bin"]) + sc = data.get("scenarios_params", {}) + for f in vars(cfg.params): + if f in sc: + setattr(cfg.params, f, sc[f]) + return cfg diff --git a/bench/engines.py b/bench/engines.py new file mode 100644 index 0000000..2592bc0 --- /dev/null +++ b/bench/engines.py @@ -0,0 +1,288 @@ +"""Engine adapters: detect, launch, await readiness, stop. + +One adapter per engine. Launch commands are built here but every argv can be +overridden from TOML (`[engines.] cmd = [...]` with {model}/{port}/... +placeholders), so a drifted upstream CLI is a config edit, not a code change. + +Servers run one at a time (Metal contention) — the orchestrator enforces that; +this module guarantees a launched process is dead before returning from stop(). +""" + +from __future__ import annotations + +import importlib.metadata +import json +import os +import shutil +import signal +import socket +import subprocess +import time +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from pathlib import Path + +from config import BenchConfig, ModelSpec, REPO_ROOT + + +@dataclass +class ServerOpts: + """Per-launch server configuration (the mlxforge config sweep varies these).""" + + label: str = "default" + kv_bits: int = 0 + prefix_cache: bool = False + max_concurrency: int = 16 # sizes llama-server's -np slots + ctx_per_slot: int = 0 # llama-server per-slot context; 0 => cfg.max_ctx + + def as_dict(self) -> dict: + return {"label": self.label, "kv_bits": self.kv_bits, "prefix_cache": self.prefix_cache, + "max_concurrency": self.max_concurrency, "ctx_per_slot": self.ctx_per_slot} + + +@dataclass +class EngineProc: + proc: subprocess.Popen + port: int + log_path: Path + + @property + def base_url(self) -> str: + return f"http://127.0.0.1:{self.port}" + + +def free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _http_get_json(url: str, timeout: float = 2.0) -> dict | None: + try: + with urllib.request.urlopen(url, timeout=timeout) as resp: + return json.loads(resp.read()) + except (urllib.error.URLError, OSError, ValueError): + return None + + +@dataclass +class EngineAdapter: + name: str + cfg: BenchConfig + skip_note: str = "" + _version: str | None = field(default=None, repr=False) + + # --- per-engine surface ------------------------------------------------- + def detect(self) -> str | None: + """Version string when usable, None to skip (set self.skip_note).""" + raise NotImplementedError + + def launch_cmd(self, model: ModelSpec, port: int, opts: ServerOpts) -> list[str]: + raise NotImplementedError + + def ready_url(self, port: int) -> str: + return f"http://127.0.0.1:{port}/v1/models" + + def is_ready(self, body: dict, model: ModelSpec) -> bool: + return True # a 200 with parseable JSON is enough by default + + def api_model_name(self, model: ModelSpec) -> str: + return model.mlx + + def extra_request_fields(self, opts: ServerOpts) -> dict: + return {} + + def health_metrics(self, port: int) -> dict | None: + return None # mlxforge only + + # --- shared mechanics ---------------------------------------------------- + def _format_cmd(self, default: list[str], **subs) -> list[str]: + template = self.cfg.engine_cmds.get(self.name, default) + return [a.format(**subs) for a in template] + + def launch(self, model: ModelSpec, opts: ServerOpts, log_dir: Path) -> EngineProc: + port = free_port() + cmd = self.launch_cmd(model, port, opts) + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / f"{self.name}-{model.key}-{opts.label}.log" + log = open(log_path, "wb") + log.write((" ".join(cmd) + "\n\n").encode()) + log.flush() + proc = subprocess.Popen( + cmd, stdout=log, stderr=subprocess.STDOUT, start_new_session=True, + ) + return EngineProc(proc=proc, port=port, log_path=log_path) + + def wait_ready(self, ep: EngineProc, model: ModelSpec, timeout: float) -> None: + deadline = time.monotonic() + timeout + url = self.ready_url(ep.port) + while time.monotonic() < deadline: + if ep.proc.poll() is not None: + raise RuntimeError( + f"{self.name} exited with code {ep.proc.returncode} during startup " + f"(see {ep.log_path})" + ) + body = _http_get_json(url) + if body is not None and self.is_ready(body, model): + return + time.sleep(0.5) + raise TimeoutError(f"{self.name} not ready after {timeout:.0f}s (see {ep.log_path})") + + def stop(self, ep: EngineProc) -> None: + # SIGINT first (mlxforge drains in-flight requests on it), then escalate. + # Signals go to the process group: some servers fork helpers. + for sig, grace in ((signal.SIGINT, 15.0), (signal.SIGTERM, 5.0), (signal.SIGKILL, 5.0)): + if ep.proc.poll() is not None: + break + try: + os.killpg(os.getpgid(ep.proc.pid), sig) + except ProcessLookupError: + break + try: + ep.proc.wait(timeout=grace) + break + except subprocess.TimeoutExpired: + continue + if ep.proc.poll() is None: + raise RuntimeError(f"{self.name} (pid {ep.proc.pid}) survived SIGKILL") + + +class MlxforgeAdapter(EngineAdapter): + def __init__(self, cfg: BenchConfig): + super().__init__(name="mlxforge", cfg=cfg) + + def detect(self) -> str | None: + if not self.cfg.mlxforge_bin.exists(): + self.skip_note = f"server binary not found at {self.cfg.mlxforge_bin} (build it first)" + return None + out = subprocess.run( + ["git", "-C", str(REPO_ROOT), "describe", "--tags", "--always", "--dirty"], + capture_output=True, text=True, + ) + return out.stdout.strip() or "unknown" + + def launch_cmd(self, model: ModelSpec, port: int, opts: ServerOpts) -> list[str]: + return self._format_cmd( + [ + str(self.cfg.mlxforge_bin), "-m", "{model}", "--host", "127.0.0.1", + "--port", "{port}", "--max-ctx", "{ctx}", + "--kv-bits", "{kv_bits}", "--prefix-cache", "{prefix_cache}", + ], + model=model.mlx, port=port, ctx=self.cfg.max_ctx, + kv_bits=opts.kv_bits, prefix_cache=int(opts.prefix_cache), + ) + + def ready_url(self, port: int) -> str: + return f"http://127.0.0.1:{port}/health" + + def is_ready(self, body: dict, model: ModelSpec) -> bool: + return body.get("status") == "ok" + + def health_metrics(self, port: int) -> dict | None: + return _http_get_json(f"http://127.0.0.1:{port}/health") + + +class LlamaCppAdapter(EngineAdapter): + def __init__(self, cfg: BenchConfig): + super().__init__(name="llamacpp", cfg=cfg) + + def detect(self) -> str | None: + if shutil.which("llama-server") is None: + self.skip_note = "llama-server not on PATH (brew install llama.cpp)" + return None + out = subprocess.run(["llama-server", "--version"], capture_output=True, text=True) + lines = ((out.stderr or "") + (out.stdout or "")).strip().splitlines() + # Backend-loading chatter precedes the "version: NNNN (sha)" line. + for line in lines: + if line.startswith("version"): + return line.split(":", 1)[1].strip() + return lines[0] if lines else "unknown" + + def launch_cmd(self, model: ModelSpec, port: int, opts: ServerOpts) -> list[str]: + np = opts.max_concurrency + model_args = ["-m", model.gguf_path] if model.gguf_path else ["-hf", model.gguf] + # -c is the *total* context, split across -np parallel slots — and the + # whole KV buffer is preallocated, so slots must be right-sized per + # launch (16 slots × 16k ctx would be tens of GB on small machines). + slot_ctx = opts.ctx_per_slot or self.cfg.max_ctx + return self._format_cmd( + [ + "llama-server", *model_args, "--host", "127.0.0.1", "--port", "{port}", + "-c", "{total_ctx}", "-np", "{np}", "-ngl", "99", "--no-webui", + ], + model=model.gguf, port=port, ctx=self.cfg.max_ctx, + total_ctx=slot_ctx * np, np=np, + ) + + def ready_url(self, port: int) -> str: + return f"http://127.0.0.1:{port}/health" + + def api_model_name(self, model: ModelSpec) -> str: + return model.gguf # llama-server serves one model; the field is echoed, not matched + + def extra_request_fields(self, opts: ServerOpts) -> dict: + # cache_prompt is the default in current builds but pinned for safety; + # ignore_eos keeps fixed-length decodes comparable across engines. + return {"ignore_eos": True, "cache_prompt": True} + + +class VllmMlxAdapter(EngineAdapter): + def __init__(self, cfg: BenchConfig): + super().__init__(name="vllm-mlx", cfg=cfg) + + def detect(self) -> str | None: + if shutil.which("vllm-mlx") is None: + self.skip_note = "vllm-mlx not on PATH (pip install vllm-mlx)" + return None + try: + return importlib.metadata.version("vllm-mlx") + except importlib.metadata.PackageNotFoundError: + return "unknown" + + def launch_cmd(self, model: ModelSpec, port: int, opts: ServerOpts) -> list[str]: + # Assumed invocation per the vllm-mlx README; override via TOML if it drifts. + return self._format_cmd( + ["vllm-mlx", "serve", "{model}", "--host", "127.0.0.1", "--port", "{port}", + "--continuous-batching"], + model=model.mlx, port=port, ctx=self.cfg.max_ctx, + ) + + +class OmlxAdapter(EngineAdapter): + def __init__(self, cfg: BenchConfig): + super().__init__(name="omlx", cfg=cfg) + + def detect(self) -> str | None: + if shutil.which("omlx") is None: + self.skip_note = "omlx not on PATH" + return None + return "unknown" + + def launch_cmd(self, model: ModelSpec, port: int, opts: ServerOpts) -> list[str]: + # omlx discovers models from subdirectories of --model-dir and selects + # per-request via the "model" field. Assumed invocation; TOML-overridable. + cmd = ["omlx", "serve", "--host", "127.0.0.1", "--port", "{port}"] + if self.cfg.omlx_model_dir: + cmd += ["--model-dir", "{model_dir}"] + return self._format_cmd( + cmd, model=model.omlx_model, port=port, model_dir=self.cfg.omlx_model_dir, + ) + + def is_ready(self, body: dict, model: ModelSpec) -> bool: + # The model list is final once /v1/models answers; a missing model means + # it isn't in the model-dir, so fail fast instead of timing out. + ids = [m.get("id", "") for m in body.get("data", [])] + if not any(model.omlx_model in i for i in ids): + raise RuntimeError( + f"omlx is up but '{model.omlx_model}' is not in its model-dir (found: {ids})" + ) + return True + + def api_model_name(self, model: ModelSpec) -> str: + return model.omlx_model + + +def make_adapters(cfg: BenchConfig) -> dict[str, EngineAdapter]: + adapters = [MlxforgeAdapter(cfg), LlamaCppAdapter(cfg), VllmMlxAdapter(cfg), OmlxAdapter(cfg)] + return {a.name: a for a in adapters} diff --git a/bench/loadgen.py b/bench/loadgen.py new file mode 100644 index 0000000..6a13805 --- /dev/null +++ b/bench/loadgen.py @@ -0,0 +1,227 @@ +"""Async load generation over OpenAI-compatible /v1/chat/completions SSE. + +Token accounting is deliberately dual: `client_chunks` (content deltas seen on +the wire) is always recorded, `server_*_tokens` comes from `usage` when the +engine reports it (llama.cpp/vLLM honor stream_options.include_usage; mlxforge +streams no usage chunk). Aggregates prefer the server count and fall back to +chunks; the report labels which one each cell used. +""" + +from __future__ import annotations + +import asyncio +import json +import statistics +import time +from dataclasses import dataclass, field + +import httpx + + +@dataclass +class RequestResult: + ok: bool = False + error: str = "" + start: float = 0.0 + first_token_at: float | None = None + end: float = 0.0 + client_chunks: int = 0 + server_prompt_tokens: int | None = None + server_completion_tokens: int | None = None + finish_reason: str = "" + saw_done: bool = False # [DONE] sentinel reached + quirk: str = "" # transport error after logical completion (tolerated) + + @property + def ttft_s(self) -> float | None: + return (self.first_token_at - self.start) if self.first_token_at else None + + @property + def latency_s(self) -> float: + return self.end - self.start + + @property + def completion_tokens(self) -> int: + return self.server_completion_tokens or self.client_chunks + + @property + def token_source(self) -> str: + return "server" if self.server_completion_tokens else "client" + + @property + def decode_tps(self) -> float | None: + # First token excluded: it belongs to prefill (TTFT), not decode. + if not self.first_token_at or self.completion_tokens < 2: + return None + dt = self.end - self.first_token_at + return (self.completion_tokens - 1) / dt if dt > 0 else None + + +def sampling_fields(max_tokens: int) -> dict: + return { + "temperature": 0, + "top_p": 1, + "seed": 1234, + "max_tokens": max_tokens, + "stream": True, + "stream_options": {"include_usage": True}, + } + + +async def stream_chat( + client: httpx.AsyncClient, + base_url: str, + model: str, + messages: list[dict], + max_tokens: int, + extra: dict | None = None, +) -> RequestResult: + body = {"model": model, "messages": messages, **sampling_fields(max_tokens), **(extra or {})} + r = RequestResult(start=time.monotonic()) + try: + async with client.stream("POST", f"{base_url}/v1/chat/completions", json=body) as resp: + if resp.status_code != 200: + detail = (await resp.aread())[:200] + r.error = f"HTTP {resp.status_code}: {detail.decode(errors='replace')}" + r.end = time.monotonic() + return r + buf = b"" + async for chunk in resp.aiter_bytes(): + buf += chunk + # SSE events are blank-line separated; engines vary on \r\n. + while (sep := _event_end(buf)) is not None: + event, buf = buf[: sep[0]], buf[sep[1]:] + _consume_event(event, r) + r.ok = not r.error + except (httpx.HTTPError, OSError) as e: + # A transport error after the stream logically completed (we saw [DONE] + # or a finish_reason) is an engine quirk worth noting, not a failed + # request — e.g. closing the connection without the chunked terminator. + if r.saw_done or r.finish_reason: + r.ok = True + r.quirk = f"{type(e).__name__} after stream end: {e}" + else: + r.error = f"{type(e).__name__}: {e}" + r.end = time.monotonic() + return r + + +def _event_end(buf: bytes) -> tuple[int, int] | None: + for sep in (b"\r\n\r\n", b"\n\n"): + i = buf.find(sep) + if i != -1: + return (i, i + len(sep)) + return None + + +def _consume_event(event: bytes, r: RequestResult) -> None: + for line in event.splitlines(): + if not line.startswith(b"data:"): + continue + data = line[5:].strip() + if data == b"[DONE]": + r.saw_done = True + return + try: + obj = json.loads(data) + except ValueError: + continue + usage = obj.get("usage") + if usage: # usage-only final chunks have empty `choices` + r.server_prompt_tokens = usage.get("prompt_tokens", r.server_prompt_tokens) + r.server_completion_tokens = usage.get("completion_tokens", r.server_completion_tokens) + for choice in obj.get("choices", []): + if choice.get("finish_reason"): + r.finish_reason = choice["finish_reason"] + delta = choice.get("delta", {}) + # reasoning_content counts too: thinking models (Qwen3) emit it as + # separate deltas on some engines, and those are generated tokens. + if delta.get("content") or delta.get("reasoning_content"): + r.client_chunks += 1 + if r.first_token_at is None: + r.first_token_at = time.monotonic() + + +@dataclass +class BatchResult: + results: list[RequestResult] = field(default_factory=list) + wall_s: float = 0.0 + + @property + def succeeded(self) -> list[RequestResult]: + return [r for r in self.results if r.ok] + + @property + def total_completion_tokens(self) -> int: + return sum(r.completion_tokens for r in self.succeeded) + + @property + def aggregate_tps(self) -> float: + return self.total_completion_tokens / self.wall_s if self.wall_s > 0 else 0.0 + + def percentile(self, attr: str, q: float) -> float | None: + vals = sorted(v for r in self.succeeded if (v := getattr(r, attr)) is not None) + if not vals: + return None + return vals[min(len(vals) - 1, int(q * len(vals)))] + + def mean(self, attr: str) -> float | None: + vals = [v for r in self.succeeded if (v := getattr(r, attr)) is not None] + return statistics.fmean(vals) if vals else None + + @property + def token_source(self) -> str: + sources = {r.token_source for r in self.succeeded} + return sources.pop() if len(sources) == 1 else "mixed" + + @property + def finish_reasons(self) -> str: + counts: dict[str, int] = {} + for r in self.succeeded: + reason = r.finish_reason or "?" + counts[reason] = counts.get(reason, 0) + 1 + return ",".join(f"{k}:{v}" for k, v in sorted(counts.items())) + + +async def run_batch( + client: httpx.AsyncClient, + base_url: str, + model: str, + message_lists: list[list[dict]], + concurrency: int, + max_tokens: int, + extra: dict | None = None, +) -> BatchResult: + sem = asyncio.Semaphore(concurrency) + + async def one(messages: list[dict]) -> RequestResult: + async with sem: + return await stream_chat(client, base_url, model, messages, max_tokens, extra) + + t0 = time.monotonic() + results = await asyncio.gather(*(one(m) for m in message_lists)) + return BatchResult(results=list(results), wall_s=time.monotonic() - t0) + + +def make_client() -> httpx.AsyncClient: + # Long read timeout: an 8k-token prefill on a small GPU can take a while. + return httpx.AsyncClient( + timeout=httpx.Timeout(connect=5.0, read=300.0, write=30.0, pool=300.0), + limits=httpx.Limits(max_connections=64), + ) + + +async def probe_prompt_tokens( + client: httpx.AsyncClient, base_url: str, model: str, text: str, extra: dict | None = None +) -> int: + """Non-streaming 1-token request; every engine reports usage here.""" + body = { + "model": model, + "messages": [{"role": "user", "content": text}], + "max_tokens": 1, + "temperature": 0, + **(extra or {}), + } + resp = await client.post(f"{base_url}/v1/chat/completions", json=body) + resp.raise_for_status() + return int(resp.json()["usage"]["prompt_tokens"]) diff --git a/bench/prompts.py b/bench/prompts.py new file mode 100644 index 0000000..6664691 --- /dev/null +++ b/bench/prompts.py @@ -0,0 +1,73 @@ +"""Deterministic, tokenizer-free prompt synthesis. + +Prompts are seeded word streams sized in *words*; a per-server two-point +calibration (one short and one long non-streaming probe, reading +`usage.prompt_tokens`) solves tokens-per-word and the fixed chat-template +overhead for that engine's tokenizer, so target token counts land within a few +percent without any Python tokenizer dependency. The achieved +`usage.prompt_tokens` is recorded per request anyway, so residual error is +visible, not silent. +""" + +from __future__ import annotations + +import random +from dataclasses import dataclass + +# A fixed mid-frequency wordlist: common enough to tokenize predictably, varied +# enough that synthesized prompts don't collapse into repeated token runs. +_WORDS = ( + "river mountain copper lantern harvest meadow timber anchor compass marble " + "garden whistle saddle barrel craft ribbon hammer voyage candle orchard " + "bridge falcon cellar pebble drift canvas mirror saddle copper village " + "window timber harbor meadow signal lantern basket runner copper anchor " + "quarry beacon cobble thicket paddle vessel slate ember willow granite" +).split() + +# The task framing is open-ended on purpose: engines should hit max_tokens, not +# EOS, so throughput numbers compare like with like (llama.cpp additionally +# gets ignore_eos from its adapter). +_TASK = ( + "Continue this never-ending inventory list, one numbered item per line, " + "without ever stopping or concluding. Reference words: " +) + + +def synth_words(n_words: int, seed: int) -> str: + rng = random.Random(seed) + return " ".join(rng.choice(_WORDS) for _ in range(n_words)) + + +@dataclass(frozen=True) +class Calibration: + tokens_per_word: float + overhead_tokens: float # chat template + task framing, in tokens + + def words_for(self, target_tokens: int) -> int: + return max(8, round((target_tokens - self.overhead_tokens) / self.tokens_per_word)) + + +def calibration_from_probes(w1: int, t1: int, w2: int, t2: int) -> Calibration: + # Two probes of w1 < w2 words measured at t1/t2 prompt tokens: the slope is + # tokens-per-word, the intercept is the fixed per-request overhead. + tpw = (t2 - t1) / (w2 - w1) + if tpw <= 0: # degenerate usage reporting; fall back to ~1.3 tokens/word + return Calibration(tokens_per_word=1.3, overhead_tokens=30.0) + return Calibration(tokens_per_word=tpw, overhead_tokens=max(0.0, t1 - w1 * tpw)) + + +CALIBRATION_PROBE_WORDS = (100, 400) + + +def probe_text(n_words: int, seed: int = 99) -> str: + return _TASK + synth_words(n_words, seed) + + +def synth_prompt(target_tokens: int, cal: Calibration, seed: int) -> str: + return _TASK + synth_words(cal.words_for(target_tokens), seed) + + +def unique_prefix(seed: int, cal: Calibration, tokens: int = 16) -> str: + # Per-request unique lead-in so prompt-prefix caches can't serve scenario + # traffic that is meant to measure prefill. + return f"(session {seed}) " + synth_words(cal.words_for(tokens), seed) + " " diff --git a/bench/report.py b/bench/report.py new file mode 100644 index 0000000..42034be --- /dev/null +++ b/bench/report.py @@ -0,0 +1,262 @@ +"""Report rendering (HTML file + text for stdout) and raw JSON persistence. + +The JSON is the source of truth (`--rerender` round-trips it); the renders are +views: one table per scenario per model, engines × levels as rows, plus an +environment header and a Notes section for skips and comparability caveats. +The saved report is a self-contained HTML file (inline CSS, no assets); the +text render is what goes to stdout. +""" + +from __future__ import annotations + +import html +import json +import subprocess +from datetime import datetime +from pathlib import Path + +# Columns are rendered in this order when present; anything else (minus _raw) +# is appended alphabetically so new scenario fields show up without edits here. +_PREFERRED_ORDER = [ + "concurrency", "requests", "prompt_tokens_target", "prompt_tokens_achieved", + "pass", "turns", "kv_bits", "prefix_cache", + "agg_tps", "prefill_tps", "decode_tps_mean", + "ttft_p50_ms", "ttft_p95_ms", "ttft_mean_ms", "ttft_min_ms", "ttft_max_ms", + "cold_ttft_ms", "warm_ttft_mean_ms", "warm_ttft_ms", "warm_speedup", + "latency_p50_s", "mean_completion_tokens", + "decode_steps", "peak_batch", "prefix_hits", "tokens_reused", "srv_avg_tps", + "errors", "token_source", "finish_reasons", "error", +] + +# Cross-row comparison direction, for best-value highlighting in the HTML. +_HIGHER_BETTER = {"agg_tps", "prefill_tps", "decode_tps_mean", "srv_avg_tps", "warm_speedup"} +_LOWER_BETTER = {"ttft_p50_ms", "ttft_p95_ms", "ttft_mean_ms", "cold_ttft_ms", + "warm_ttft_mean_ms", "warm_ttft_ms", "latency_p50_s"} + +_SCENARIO_TITLES = { + "concurrency": "Concurrency sweep", + "prompt": "Prompt-length sweep (single stream)", + "config": "mlxforge config sweep (kv_bits × prefix_cache)", + "multiturn": "Multi-turn shared-prefix reuse", +} + + +def _sysctl(key: str) -> str: + out = subprocess.run(["sysctl", "-n", key], capture_output=True, text=True) + return out.stdout.strip() + + +def host_info() -> dict: + sw = subprocess.run(["sw_vers", "-productVersion"], capture_output=True, text=True) + mem_bytes = int(_sysctl("hw.memsize") or 0) + return { + "model": _sysctl("hw.model"), + "chip": _sysctl("machdep.cpu.brand_string"), + "ram_gb": round(mem_bytes / 2**30), + "macos": sw.stdout.strip(), + } + + +def _fmt_cell(v) -> str: + if v is None: + return "-" + if isinstance(v, float): + return f"{v:g}" + return str(v) + + +# --- shared table assembly ---------------------------------------------------- + +def _table_columns(rows: list[dict], lead_cols: list[str]) -> list[str]: + keys: list[str] = list(lead_cols) + seen = set(keys) + for col in _PREFERRED_ORDER: + if col not in seen and any(col in r for r in rows): + keys.append(col) + seen.add(col) + for r in rows: + for col in sorted(r): + if col not in seen and not col.startswith("_"): + keys.append(col) + seen.add(col) + return keys + + +def _grouped_tables(results: dict) -> list[tuple[str, str, str, list[dict]]]: + """(scenario, model, subtitle, rows) per table, in scenario order; each row + carries an `engine` lead column derived from the run's server-config label.""" + groups: dict[tuple[str, str], list[dict]] = {} + for run in results["runs"]: + if run.get("rows"): + groups.setdefault((run["scenario"], run["model"]), []).append(run) + tables = [] + for scenario in _SCENARIO_TITLES: + for (sc, model), group in groups.items(): + if sc != scenario: + continue + rows = [] + for run in group: + label = run["server_config"].get("label", "default") + engine = run["engine"] if label == "default" else f"{run['engine']}[{label}]" + for row in run["rows"]: + rows.append({"engine": engine, + **{k: v for k, v in row.items() if not k.startswith("_")}}) + tables.append((scenario, model, _scenario_subtitle(sc, group[0].get("params", {})), rows)) + return tables + + +def _notes(results: dict) -> list[str]: + notes = [f"{name} skipped: {info['note']}" + for name, info in results["engines"].items() if not info.get("detected")] + notes += [f"{r['engine']}/{r['model']}: {r['note']}" for r in results["runs"] if r.get("note")] + notes += [ + "token_source: 'server' = engine-reported usage; 'client' = SSE content-chunk count.", + "GGUF quants (e.g. Q4_K_M) are nearest equivalents to MLX group quants, not identical.", + ] + return notes + + +def _engines_line(results: dict) -> str: + return " · ".join( + f"{name} {info['version']}" if info.get("detected") else f"{name} SKIPPED" + for name, info in results["engines"].items()) + + +def _scenario_subtitle(scenario: str, p: dict) -> str: + if scenario == "concurrency": + return (f"prompt≈{p.get('concurrency_prompt_tokens')} tok" + f" · max_tokens={p.get('concurrency_max_tokens')}") + if scenario == "prompt": + return f"max_tokens={p.get('prompt_max_tokens')}" + if scenario == "multiturn": + return (f"system≈{p.get('multiturn_system_tokens')} tok" + f" · {p.get('multiturn_turns')} turns") + return "" + + +# --- text render (stdout) ------------------------------------------------------ + +def format_table(rows: list[dict], lead_cols: list[str]) -> str: + keys = _table_columns(rows, lead_cols) + table = [keys] + [[_fmt_cell(r.get(k)) for k in keys] for r in rows] + widths = [max(len(row[i]) for row in table) for i in range(len(keys))] + lines = [" ".join(c.ljust(w) for c, w in zip(row, widths)).rstrip() for row in table] + lines.insert(1, " ".join("-" * w for w in widths)) + return "\n".join(lines) + + +def render(results: dict) -> str: + meta = results["meta"] + h = meta.get("host", {}) + out = [ + f"mlxforge benchmark harness — {meta.get('timestamp', '')}", + f"Host: {h.get('model', '?')} · {h.get('chip', '?')} · {h.get('ram_gb', '?')} GB · " + f"macOS {h.get('macos', '?')}", + "Engines: " + _engines_line(results), + "", + ] + for scenario, model, subtitle, rows in _grouped_tables(results): + sub = f" · {subtitle}" if subtitle else "" + out.append(f"== {_SCENARIO_TITLES[scenario]} · {model}{sub} ==") + out.append(format_table(rows, lead_cols=["engine"])) + out.append("") + out.append("Notes:") + out += [f"- {n}" for n in _notes(results)] + return "\n".join(out) + "\n" + + +# --- HTML render (the report file) --------------------------------------------- + +_CSS = """ +:root { color-scheme: light dark; } +body { font: 15px/1.5 -apple-system, "Helvetica Neue", sans-serif; margin: 2rem auto; + max-width: 72rem; padding: 0 1rem; } +h1 { font-size: 1.4rem; margin-bottom: .2rem; } +h2 { font-size: 1.1rem; margin: 2rem 0 .2rem; } +.sub { color: #777; font-size: .85rem; margin: 0 0 .6rem; } +table { border-collapse: collapse; width: 100%; font-size: .85rem; } +th, td { padding: .35rem .6rem; text-align: right; white-space: nowrap; } +th:first-child, td:first-child { text-align: left; } +th { border-bottom: 2px solid #888; font-weight: 600; } +td { border-bottom: 1px solid rgba(136,136,136,.25); + font-variant-numeric: tabular-nums; } +tr:hover td { background: rgba(136,136,136,.08); } +td.best { font-weight: 700; color: #0a7d33; } +@media (prefers-color-scheme: dark) { td.best { color: #4cc36e; } } +ul.notes { color: #777; font-size: .85rem; } +""" + + +# Rows are only comparable at the same sweep point (same concurrency level, +# prompt size, ...); highlighting is computed per group of equal key value. +# The config sweep has no group key: every row is a config, all comparable. +_GROUP_KEY = {"concurrency": "concurrency", "prompt": "prompt_tokens_target", + "multiturn": "pass"} + + +def _best_values(rows: list[dict], keys: list[str], scenario: str) -> dict[tuple, float]: + group_key = _GROUP_KEY.get(scenario) + groups: dict[object, list[dict]] = {} + for r in rows: + groups.setdefault(r.get(group_key) if group_key else None, []).append(r) + best: dict[tuple, float] = {} + for gval, grows in groups.items(): + if len(grows) < 2: + continue + for k in keys: + vals = [r[k] for r in grows if isinstance(r.get(k), (int, float))] + if len(vals) < 2 or min(vals) == max(vals): + continue + if k in _HIGHER_BETTER: + best[(gval, k)] = max(vals) + elif k in _LOWER_BETTER: + best[(gval, k)] = min(vals) + return best + + +def render_html(results: dict) -> str: + meta = results["meta"] + h = meta.get("host", {}) + e = html.escape + out = [ + "", + "mlxforge benchmark report", + f"", + f"

mlxforge benchmark report

", + f"

{e(meta.get('timestamp', ''))}" + f"{' · quick run' if meta.get('quick') else ''}
" + f"Host: {e(h.get('model', '?'))} · {e(h.get('chip', '?'))} · " + f"{e(str(h.get('ram_gb', '?')))} GB · macOS {e(h.get('macos', '?'))}
" + f"Engines: {e(_engines_line(results))}

", + ] + for scenario, model, subtitle, rows in _grouped_tables(results): + keys = _table_columns(rows, ["engine"]) + best = _best_values(rows, keys, scenario) + group_key = _GROUP_KEY.get(scenario) + out.append(f"

{e(_SCENARIO_TITLES[scenario])} · {e(model)}

") + if subtitle: + out.append(f"

{e(subtitle)}

") + out.append("" + "".join(f"" for k in keys) + "") + for r in rows: + gval = r.get(group_key) if group_key else None + cells = [] + for k in keys: + v = r.get(k) + cls = " class='best'" if v is not None and best.get((gval, k)) == v else "" + cells.append(f"{e(_fmt_cell(v))}") + out.append("" + "".join(cells) + "") + out.append("
{e(k)}
") + out.append("

Notes

    ") + out += [f"
  • {e(n)}
    • " for n in _notes(results)] + out.append("
") + return "\n".join(out) + "\n" + + +def save(results: dict, out_dir: Path) -> tuple[Path, Path]: + out_dir.mkdir(parents=True, exist_ok=True) + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + json_path = out_dir / f"bench-{ts}.json" + html_path = out_dir / f"bench-{ts}.html" + json_path.write_text(json.dumps(results, indent=2, default=str)) + html_path.write_text(render_html(results)) + return html_path, json_path diff --git a/bench/scenarios.py b/bench/scenarios.py new file mode 100644 index 0000000..30d96a9 --- /dev/null +++ b/bench/scenarios.py @@ -0,0 +1,261 @@ +"""The four benchmark scenarios. + +Each runner takes a ServerCtx (a live, warmed, calibrated server) and returns +a list of row dicts that report.py renders verbatim; raw per-request data +rides along in the JSON for re-analysis. + +Prompt hygiene: scenarios 1–2 give every request a unique lead-in so engines +with prompt/prefix caches measure real prefill; scenario 4 does the opposite +on purpose (shared system prompt) to measure those caches. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field + +import httpx + +import loadgen +import prompts +from config import ScenarioParams +from engines import EngineAdapter, EngineProc, ServerOpts + + +@dataclass +class ServerCtx: + adapter: EngineAdapter + ep: EngineProc + model_name: str # the "model" field requests must carry + opts: ServerOpts + client: httpx.AsyncClient + extra: dict = field(default_factory=dict) + # Set right after launch by calibrate(); every scenario depends on it. + cal: prompts.Calibration = field(default=None) # type: ignore[assignment] + + def health(self) -> dict | None: + return self.adapter.health_metrics(self.ep.port) + + +async def calibrate(ctx: ServerCtx) -> prompts.Calibration: + w1, w2 = prompts.CALIBRATION_PROBE_WORDS + t1 = await loadgen.probe_prompt_tokens( + ctx.client, ctx.ep.base_url, ctx.model_name, prompts.probe_text(w1), ctx.extra) + t2 = await loadgen.probe_prompt_tokens( + ctx.client, ctx.ep.base_url, ctx.model_name, prompts.probe_text(w2), ctx.extra) + return prompts.calibration_from_probes(w1, t1, w2, t2) + + +async def warmup(ctx: ServerCtx) -> None: + # Two discarded generations cover Metal JIT / first-graph compilation. + msgs = [[{"role": "user", "content": prompts.synth_prompt(256, ctx.cal, seed=7 + i)}] + for i in range(2)] + await loadgen.run_batch(ctx.client, ctx.ep.base_url, ctx.model_name, msgs, + concurrency=1, max_tokens=32, extra=ctx.extra) + + +def _request_raw(r: loadgen.RequestResult) -> dict: + return { + "ok": r.ok, "error": r.error, "ttft_s": r.ttft_s, "latency_s": r.latency_s, + "client_chunks": r.client_chunks, "server_prompt_tokens": r.server_prompt_tokens, + "server_completion_tokens": r.server_completion_tokens, + "finish_reason": r.finish_reason, "quirk": r.quirk, + } + + +def _batch_raw(b: loadgen.BatchResult) -> dict: + return {"wall_s": b.wall_s, "requests": [_request_raw(r) for r in b.results]} + + +async def run_concurrency(ctx: ServerCtx, p: ScenarioParams, quick: bool) -> list[dict]: + rows = [] + for i, level in enumerate(p.concurrency_levels): + total = max(8, 2 * level) if not quick else level + # Repeat the first level once and keep the second pass: the very first + # batch after warmup still absorbs residual one-time costs. Each pass + # gets fresh prompts — reusing them would let engines with prompt + # caches (llama.cpp cache_prompt) skip the kept pass's prefill. + passes = 1 if (quick or i > 0) else 2 + for pass_i in range(passes): + msgs = [] + for j in range(total): + seed = 1000 * level + j + 500_000 * pass_i + body = prompts.synth_prompt(p.concurrency_prompt_tokens, ctx.cal, seed=seed) + msgs.append( + [{"role": "user", "content": prompts.unique_prefix(seed, ctx.cal) + body}]) + batch = await loadgen.run_batch( + ctx.client, ctx.ep.base_url, ctx.model_name, msgs, + concurrency=level, max_tokens=p.concurrency_max_tokens, extra=ctx.extra) + errors = len(batch.results) - len(batch.succeeded) + rows.append({ + "concurrency": level, + "requests": total, + "agg_tps": round(batch.aggregate_tps, 1), + "ttft_p50_ms": _ms(batch.percentile("ttft_s", 0.50)), + "ttft_p95_ms": _ms(batch.percentile("ttft_s", 0.95)), + "decode_tps_mean": _r1(batch.mean("decode_tps")), + "latency_p50_s": _r2(batch.percentile("latency_s", 0.50)), + "mean_completion_tokens": _r1(batch.mean("completion_tokens")), + "errors": errors, + "token_source": batch.token_source, + "finish_reasons": batch.finish_reasons, + "_raw": _batch_raw(batch), + }) + return rows + + +async def run_prompt_sweep(ctx: ServerCtx, p: ScenarioParams, quick: bool) -> list[dict]: + rows = [] + for size in p.prompt_sizes: + ttfts, ptoks, raws = [], [], [] + for rep in range(p.prompt_repeats): + seed = 10_000 + size + rep + text = (prompts.unique_prefix(seed, ctx.cal) + + prompts.synth_prompt(size, ctx.cal, seed=seed)) + before = ctx.health() + batch = await loadgen.run_batch( + ctx.client, ctx.ep.base_url, ctx.model_name, + [[{"role": "user", "content": text}]], + concurrency=1, max_tokens=p.prompt_max_tokens, extra=ctx.extra) + after = ctx.health() + raws.append(_batch_raw(batch)) + for r in batch.succeeded: + if r.ttft_s is not None: + ttfts.append(r.ttft_s) + if r.server_prompt_tokens: + ptoks.append(r.server_prompt_tokens) + elif before and after: + # mlxforge streams no usage; its /health prompt-token counter + # over a single request gives the same number. + d = (after["decode"]["prompt_tokens_total"] + - before["decode"]["prompt_tokens_total"]) + if d > 0: + ptoks.append(d) + if not ttfts: + rows.append({"prompt_tokens_target": size, "error": "all requests failed", + "_raw": raws}) + continue + mean_ttft = sum(ttfts) / len(ttfts) + achieved = round(sum(ptoks) / len(ptoks)) if ptoks else None + rows.append({ + "prompt_tokens_target": size, + "prompt_tokens_achieved": achieved, + "ttft_mean_ms": _ms(mean_ttft), + "ttft_min_ms": _ms(min(ttfts)), + "ttft_max_ms": _ms(max(ttfts)), + # Prefill rate from achieved tokens when usage exists, else target. + "prefill_tps": round((achieved or size) / mean_ttft) if mean_ttft > 0 else None, + "_raw": raws, + }) + return rows + + +def make_multiturn_messages(system_tokens: int, cal: prompts.Calibration, + pass_seed: int) -> list[dict]: + system = prompts.synth_prompt(system_tokens, cal, seed=pass_seed) + return [{"role": "system", "content": "You are a meticulous inventory clerk. " + system}] + + +async def run_multiturn(ctx: ServerCtx, p: ScenarioParams, quick: bool) -> list[dict]: + rows = [] + for pass_i in range(p.multiturn_passes): + base = make_multiturn_messages(p.multiturn_system_tokens, ctx.cal, 77_000 + pass_i) + history: list[dict] = [] + cold_ttft, warm_ttfts, raws = None, [], [] + for turn in range(p.multiturn_turns): + messages = base + history + [ + {"role": "user", "content": f"Continue the list (part {turn + 1})."}] + batch = await loadgen.run_batch( + ctx.client, ctx.ep.base_url, ctx.model_name, [messages], + concurrency=1, max_tokens=p.multiturn_max_tokens, extra=ctx.extra) + raws.append(_batch_raw(batch)) + ok = batch.succeeded + if not ok or ok[0].ttft_s is None: + break + if turn == 0: + cold_ttft = ok[0].ttft_s + else: + warm_ttfts.append(ok[0].ttft_s) + # Keep history growth bounded; the shared prefix is what matters. + history += [ + {"role": "user", "content": f"Continue the list (part {turn + 1})."}, + {"role": "assistant", "content": f"(items for part {turn + 1} omitted)"}, + ] + if cold_ttft is None: + rows.append({"pass": pass_i + 1, "error": "cold turn failed", "_raw": raws}) + continue + warm = sum(warm_ttfts) / len(warm_ttfts) if warm_ttfts else None + rows.append({ + "pass": pass_i + 1, + "turns": p.multiturn_turns, + "cold_ttft_ms": _ms(cold_ttft), + "warm_ttft_mean_ms": _ms(warm), + "warm_speedup": round(cold_ttft / warm, 2) if warm else None, + "_raw": raws, + }) + return rows + + +async def run_config_probes(ctx: ServerCtx, p: ScenarioParams, quick: bool) -> list[dict]: + """mlxforge config sweep: one launch of the kv_bits×prefix matrix runs this. + + Cost probe (concurrency batch) + benefit probe (multi-turn), bracketed by + /health snapshots so server-side counters (decode steps, prefix hits) + sit beside the client-side numbers. + """ + before = ctx.health() or {} + level = p.config_probe_concurrency # already shrunk in quick mode (see ScenarioParams.quick) + msgs = [] + for j in range(max(8, level)): + seed = 50_000 + j + body = prompts.synth_prompt(p.concurrency_prompt_tokens, ctx.cal, seed=seed) + msgs.append([{"role": "user", "content": prompts.unique_prefix(seed, ctx.cal) + body}]) + cost = await loadgen.run_batch( + ctx.client, ctx.ep.base_url, ctx.model_name, msgs, + concurrency=level, max_tokens=p.concurrency_max_tokens, extra=ctx.extra) + benefit_rows = await run_multiturn(ctx, p, quick) + after = ctx.health() or {} + + def delta(*path: str) -> int | None: + def dig(d: dict): + for k in path: + d = d.get(k, {}) if isinstance(d, dict) else {} + return d if isinstance(d, (int, float)) else None + b, a = dig(before), dig(after) + return int(a - b) if a is not None and b is not None else None + + warm = [r.get("warm_ttft_mean_ms") for r in benefit_rows if r.get("warm_ttft_mean_ms")] + return [{ + "kv_bits": ctx.opts.kv_bits, + "prefix_cache": "on" if ctx.opts.prefix_cache else "off", + "agg_tps": round(cost.aggregate_tps, 1), + "ttft_p50_ms": _ms(cost.percentile("ttft_s", 0.50)), + "warm_ttft_ms": round(sum(warm) / len(warm), 1) if warm else None, + "decode_steps": delta("decode", "steps"), + "peak_batch": (after.get("batch", {}) or {}).get("peak"), + "prefix_hits": delta("prefix_cache", "hits"), + "tokens_reused": delta("prefix_cache", "tokens_reused"), + "srv_avg_tps": round((after.get("decode", {}) or {}).get("avg_tokens_per_second", 0), 1), + "_raw": {"cost": _batch_raw(cost), "benefit": benefit_rows, + "health_before": before, "health_after": after}, + }] + + +def _ms(v: float | None) -> float | None: + return round(v * 1000, 1) if v is not None else None + + +def _r1(v: float | None) -> float | None: + return round(v, 1) if v is not None else None + + +def _r2(v: float | None) -> float | None: + return round(v, 2) if v is not None else None + + +RUNNERS = { + "concurrency": run_concurrency, + "prompt": run_prompt_sweep, + "multiturn": run_multiturn, + "config": run_config_probes, +} diff --git a/doc/applications.md b/doc/applications.md index 02c6085..bcee8c1 100644 --- a/doc/applications.md +++ b/doc/applications.md @@ -282,3 +282,20 @@ any tensor is not fp16. Confirms Metal is available, adds two small arrays, calls `mx::eval`, and prints the sum — the minimal "MLX is working" check and a demonstration of MLX's lazy evaluation (nothing computes until `eval`). + +## The benchmark harness: `bench/` + +```sh +uv run bench/bench.py --quick --engines mlxforge --models qwen3-0.6b +``` + +A Python orchestrator (not a build target) that measures the engine under +different configurations — concurrency levels, prompt lengths, `kv_bits`, +prefix cache on/off, multi-turn prefix reuse — and compares it against the +other local engines on Apple Silicon (llama.cpp's `llama-server`, vllm-mlx, +omlx), all driven identically through their OpenAI-compatible streaming APIs, +one engine at a time. It launches and tears down each server itself, skips +engines that aren't installed, and writes a plain-text report plus raw JSON +under `bench/results/`. It complements the CLI's single-stream `bench` / +`bench-prefix` subcommands with the cross-engine and continuous-batching +story. See [`bench/README.md`](../bench/README.md). From cff7259ca9ec23d705623cc9e156ffed5a33a0c5 Mon Sep 17 00:00:00 2001 From: Helder Vasconcelos Date: Thu, 11 Jun 2026 19:58:43 +0100 Subject: [PATCH 3/3] bench: resolve vllm-mlx version from its own venv for pipx/uv-tool installs importlib.metadata only sees the bench environment; an entry script installed by pipx/uv tool lives in its own venv, so fall back to asking the interpreter from the script's shebang. Verified against a real vllm-mlx 0.3.0 install (all three cross-engine scenarios pass). Co-Authored-By: Claude Fable 5 --- bench/engines.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/bench/engines.py b/bench/engines.py index 2592bc0..94ecd54 100644 --- a/bench/engines.py +++ b/bench/engines.py @@ -238,7 +238,18 @@ def detect(self) -> str | None: try: return importlib.metadata.version("vllm-mlx") except importlib.metadata.PackageNotFoundError: - return "unknown" + pass + # pipx/uv-tool installs live in their own venv; ask the entry script's + # interpreter (its shebang) for the package version. + shebang = Path(shutil.which("vllm-mlx")).read_text(errors="replace").splitlines()[0] + if shebang.startswith("#!"): + out = subprocess.run( + [shebang[2:].strip(), "-c", + "import importlib.metadata as m; print(m.version('vllm-mlx'))"], + capture_output=True, text=True) + if out.returncode == 0: + return out.stdout.strip() + return "unknown" def launch_cmd(self, model: ModelSpec, port: int, opts: ServerOpts) -> list[str]: # Assumed invocation per the vllm-mlx README; override via TOML if it drifts.