From a1a973eaa41c47d218eda917eaa61d45f39957ca Mon Sep 17 00:00:00 2001 From: "Gabriel A. Devenyi" Date: Sat, 22 Aug 2026 10:45:30 -0400 Subject: [PATCH 1/3] feat(bench): `ft bench decode` -- end-to-end tokens/s, with A/B across engine flags `ft bench bw` measures one kernel in isolation over a small synthetic bank. That is the right shape for calibrating a bandwidth ratio and the wrong shape for answering "is this configuration faster", because a serving step is the kernel *plus* the gather it contends with, the KV traffic, the handshake and the scheduler, on the real model at the real cache size. They can disagree completely. Confining the CPU MoE pool to one NUMA node measured +42% (bf16) and +30% (ds_fp4) on the microbenchmark and -6.7% on tokens/s serving DeepSeek-V4-Flash; under `--moe-backend cpu` it was -28%. Answering that took a throwaway shell script -- start a server, poll for ready, curl some completions, grep gen throughput out of the log, kill it, repeat. I wrote that script three times in one afternoon, which is the actual argument for this command. ft bench decode --model DIR ft bench decode --model DIR --compare moe-backend=hybrid,offload ft bench decode --model DIR --compare moe-cache-rate=0.1,0.25,0.5 --cycles 3 Two things it does that the hand-rolled version did not: * **Prefill is subtracted.** The same prompt is timed at `max_tokens=1` and at `max_tokens=n`; `(n-1)/(t_n - t_1)` is decode alone, so prompt length cannot quietly flatter the result. `ignore_eos` keeps every run the same length. * **Variants alternate**, one full pass per cycle rather than all of A then all of B. Thermal drift and page-cache state move over minutes; blocked runs attribute that drift to the variant. A single cycle prints a warning saying so. Each measurement runs in a fresh subprocess: expert banks are pinned and there is no `cudaHostUnregister` binding, so an in-process teardown does not reliably give the memory back and the next variant measures the last one's leftovers. Validated against numbers previously obtained by hand through the HTTP server on Qwen3.6-35B-A3B-FP8 -- same ordering and magnitude, and much steadier: moe-cache-rate=0.10 40.06 tok/s (40.0-40.2) server run: 44.16 moe-cache-rate=0.25 69.48 tok/s (69.4-69.5) server run: 82.31 The offset is expected and intended: this reports steady-state decode with prefill removed, so it is a tool for comparing configurations, not for quoting a serving figure. The 0.5% spread across cycles is the point -- the server-log method scattered by several percent, which is the same order as the effects being chased. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GpGe2fQ5pDShGrnSuksfun --- python/freetoken/cli.py | 9 +- python/freetoken/moe/bench_decode.py | 211 +++++++++++++++++++++++++++ 2 files changed, 218 insertions(+), 2 deletions(-) create mode 100644 python/freetoken/moe/bench_decode.py diff --git a/python/freetoken/cli.py b/python/freetoken/cli.py index 4e6deff23..3316ce7f7 100644 --- a/python/freetoken/cli.py +++ b/python/freetoken/cli.py @@ -16,7 +16,7 @@ def _print_help(file: TextIO) -> None: daemon Run the FreeToken supervisor (persistent engine service) launch Configure and launch an agent against a FreeToken server checkpoint Convert an HF safetensors checkpoint to FTW - bench Run a micro-benchmark (e.g. "bench bw" = CPU vs PCIe bandwidth) + bench Run a benchmark ("bench bw" = bandwidth, "bench decode" = tokens/s) Use "ft --help" for command-specific options. Use "ft --version" to print the FreeToken version.""", @@ -66,7 +66,8 @@ def _print_bench_help(file: TextIO) -> None: """usage: ft bench [args] Subcommands: - bw Benchmark CPU vs PCIe bandwidth and pick the MoE backend (hybrid/offload) + bw Benchmark CPU vs PCIe bandwidth and pick the MoE backend (hybrid/offload) + decode Measure end-to-end decode tokens/s, optionally A/B across engine flags Use "ft bench --help" for subcommand-specific options.""", file=file, @@ -85,6 +86,10 @@ def _run_bench(argv: list[str]) -> int: from freetoken.moe.benchbw import main return main(argv[1:], prog="ft bench bw") + if sub == "decode": + from freetoken.moe.bench_decode import main + + return main(argv[1:], prog="ft bench decode") print(f"unknown ft bench subcommand: {sub}", file=sys.stderr) _print_bench_help(sys.stderr) return 2 diff --git a/python/freetoken/moe/bench_decode.py b/python/freetoken/moe/bench_decode.py new file mode 100644 index 000000000..d26daf429 --- /dev/null +++ b/python/freetoken/moe/bench_decode.py @@ -0,0 +1,211 @@ +"""``ft bench decode``: end-to-end decode throughput, and A/B across engine flags. + +``ft bench bw`` measures one kernel in isolation over a small synthetic bank. That is +the right shape for calibrating a bandwidth ratio and the wrong shape for answering +"is this configuration faster", because a serving step is the kernel *plus* the PCIe +gather it contends with, the KV traffic, the GPU<->CPU handshake and the scheduler -- +on the real model, at the real cache size. The two can disagree completely: on one +2-socket box a change measured +30% on the microbenchmark and -6.7% on tokens/s. + +So this loads the actual model and generates. + + ft bench decode --model DIR + ft bench decode --model DIR --compare moe-backend=hybrid,offload + ft bench decode --model DIR --compare moe-cache-rate=0.1,0.25,0.5 --cycles 3 + +Two things it does that a hand-rolled loop usually does not: + +* **Decode is isolated from prefill** by timing the same prompt twice, once with + ``max_tokens=1`` and once with ``max_tokens=n``, and taking ``(n-1)/(t_n - t_1)``. + Prefill cost cancels, so a long prompt does not quietly flatter the result. +* **Variants alternate**, one full pass per cycle rather than all runs of A then all + of B. Thermal drift, page-cache state and whatever else the box is doing move over + minutes; blocked runs attribute that drift to the variant. + +Each measurement runs in a fresh subprocess. Expert banks are pinned and cannot be +unregistered, so tearing an engine down in-process does not reliably give the memory +back -- the second variant would be measuring the first one's leftovers. +""" + +from __future__ import annotations + +import argparse +import json +import os +import statistics +import subprocess +import sys +import time + +DEFAULT_PROMPT = ( + "Write a detailed technical explanation of how a mixture-of-experts transformer " + "routes tokens to experts, and why that makes memory bandwidth the bottleneck." +) + + +def _coerce(v: str): + """CLI strings to the types SchedulerConfig fields expect.""" + low = v.strip().lower() + if low in ("true", "false"): + return low == "true" + if low in ("none", "null"): + return None + for cast in (int, float): + try: + return cast(v) + except ValueError: + pass + return v + + +def measure_decode_tps(model_path: str, engine_kwargs: dict, prompt: str, + tokens: int, samples: int) -> dict: + """Decode tokens/s for one engine configuration, median over ``samples``.""" + import torch + + from freetoken.core import SamplingParams + from freetoken.llm import LLM + + llm = LLM(model_path, dtype=torch.bfloat16, **engine_kwargs) + # `ignore_eos` so every run generates exactly `tokens` -- otherwise an early stop + # silently shortens the measurement and inflates the rate. + one = SamplingParams(max_tokens=1, temperature=0.0, ignore_eos=True) + many = SamplingParams(max_tokens=tokens, temperature=0.0, ignore_eos=True) + + llm.generate([prompt], SamplingParams(max_tokens=8, temperature=0.0, ignore_eos=True)) + + rates = [] + for _ in range(samples): + t0 = time.perf_counter() + llm.generate([prompt], one) + t_prefill = time.perf_counter() - t0 + t0 = time.perf_counter() + llm.generate([prompt], many) + t_total = time.perf_counter() - t0 + decode_s = t_total - t_prefill + if decode_s <= 0: + continue + rates.append((tokens - 1) / decode_s) + if not rates: + raise RuntimeError("no usable timing samples") + return { + "tps": statistics.median(rates), + "samples": [round(r, 2) for r in rates], + "prefill_s": round(t_prefill, 3), + } + + +def _run_worker(argv: list[str]) -> int: + """Hidden per-measurement subprocess: emit one JSON line on stdout.""" + spec = json.loads(argv[0]) + try: + out = measure_decode_tps(spec["model"], spec["kwargs"], spec["prompt"], + spec["tokens"], spec["samples"]) + except Exception as e: # noqa: BLE001 - reported to the parent, not swallowed + print(json.dumps({"error": f"{type(e).__name__}: {e}"}), flush=True) + return 1 + print(json.dumps(out), flush=True) + return 0 + + +def _measure_in_subprocess(spec: dict, quiet: bool) -> dict: + proc = subprocess.run( + [sys.executable, "-m", "freetoken.moe.bench_decode", "--_worker", json.dumps(spec)], + capture_output=True, text=True, check=False, + env={**os.environ, **spec.get("env", {})}, + ) + line = next((ln for ln in reversed(proc.stdout.splitlines()) if ln.startswith("{")), None) + if line is None: + tail = (proc.stderr or proc.stdout).strip().splitlines()[-6:] + return {"error": "no result from worker:\n " + "\n ".join(tail)} + out = json.loads(line) + if "error" in out and not quiet: + print(f" worker: {out['error']}", file=sys.stderr) + return out + + +def _variants(compare: str | None, extra: list[str]) -> list[tuple[str, dict]]: + base = {} + for kv in extra: + k, _, v = kv.partition("=") + base[k.strip().lstrip("-").replace("-", "_")] = _coerce(v) + if not compare: + return [("baseline", base)] + flag, _, values = compare.partition("=") + if not values: + raise SystemExit("--compare wants FLAG=value1,value2") + key = flag.strip().lstrip("-").replace("-", "_") + return [(f"{flag}={v}", {**base, key: _coerce(v)}) for v in values.split(",")] + + +def main(argv: list[str] | None = None, prog: str = "ft bench decode") -> int: + argv = list(sys.argv[1:] if argv is None else argv) + if argv and argv[0] == "--_worker": + return _run_worker(argv[1:]) + + p = argparse.ArgumentParser(prog=prog, description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--model", required=True, help="local checkpoint directory") + p.add_argument("--compare", default=None, metavar="FLAG=A,B", + help="engine flag to vary, e.g. 'moe-backend=hybrid,offload'") + p.add_argument("--set", action="append", default=[], metavar="FLAG=VALUE", + help="engine flag held fixed across variants (repeatable)") + p.add_argument("--cycles", type=int, default=2, + help="alternating passes over the variants (default 2)") + p.add_argument("--samples", type=int, default=3, + help="timed generations per load (default 3)") + p.add_argument("--tokens", type=int, default=128, help="tokens per generation") + p.add_argument("--prompt", default=DEFAULT_PROMPT) + p.add_argument("-o", "--out", default=None, help="write results as JSON") + p.add_argument("-q", "--quiet", action="store_true") + ns = p.parse_args(argv) + + if not os.path.isdir(ns.model): + raise SystemExit(f"--model must be a local directory (got {ns.model!r})") + + variants = _variants(ns.compare, ns.set) + results: dict[str, list[float]] = {name: [] for name, _ in variants} + print(f" {len(variants)} variant(s) x {ns.cycles} cycle(s), " + f"{ns.samples} timed generations of {ns.tokens} tokens each") + print(" each measurement reloads the model in a fresh process\n") + + for cycle in range(1, ns.cycles + 1): + for name, kwargs in variants: + spec = {"model": ns.model, "kwargs": kwargs, "prompt": ns.prompt, + "tokens": ns.tokens, "samples": ns.samples} + t0 = time.perf_counter() + out = _measure_in_subprocess(spec, ns.quiet) + dt = time.perf_counter() - t0 + if "error" in out: + print(f" cycle {cycle} {name:<28} FAILED ({dt:.0f}s)") + continue + results[name].append(out["tps"]) + print(f" cycle {cycle} {name:<28} {out['tps']:7.2f} tok/s ({dt:.0f}s)") + + print() + rows = [(n, v) for n, v in results.items() if v] + if not rows: + print(" no successful measurements") + return 1 + best = max(statistics.median(v) for _, v in rows) + print(f" {'variant':<28} {'median':>9} {'spread':>17} vs best") + for name, vals in rows: + med = statistics.median(vals) + spread = f"{min(vals):.1f}-{max(vals):.1f}" if len(vals) > 1 else "-" + rel = "best" if med == best else f"{(med / best - 1) * 100:+.1f}%" + print(f" {name:<28} {med:8.2f} {spread:>17} {rel}") + if any(len(v) < 2 for _, v in rows): + print("\n Only one cycle per variant: nothing separates a real difference from " + "drift.\n Use --cycles 2 or more before believing a small gap.") + + if ns.out: + with open(ns.out, "w") as f: + json.dump({"model": ns.model, "tokens": ns.tokens, + "results": {n: v for n, v in results.items()}}, f, indent=2) + f.write("\n") + print(f"\n saved: {ns.out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 17193b2ffdd294cfeccd1bc1e8ec2d5c6150ae8f Mon Sep 17 00:00:00 2001 From: "Gabriel A. Devenyi" Date: Sat, 22 Aug 2026 11:57:31 -0400 Subject: [PATCH 2/3] fix(bench decode): default offload-family backends to --moe-cache-auto `ft serve` defaults offload-family backends to --moe-cache-auto when no cache-sizing flag is given (prepare_server_args); the offline LLM path does not, so `--compare moe-backend=hybrid,offload` died on "moe_cache_size=0 is too small" before measuring anything. Mirror the CLI. An explicit --set moe-cache-size / -rate / -auto still wins. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GpGe2fQ5pDShGrnSuksfun --- python/freetoken/moe/bench_decode.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/python/freetoken/moe/bench_decode.py b/python/freetoken/moe/bench_decode.py index d26daf429..f8ed71e1d 100644 --- a/python/freetoken/moe/bench_decode.py +++ b/python/freetoken/moe/bench_decode.py @@ -66,6 +66,13 @@ def measure_decode_tps(model_path: str, engine_kwargs: dict, prompt: str, from freetoken.core import SamplingParams from freetoken.llm import LLM + # `ft serve` defaults the offload-family backends to --moe-cache-auto when no + # cache-sizing flag is given (prepare_server_args); the offline LLM path does not, + # so a bare `--compare moe-backend=hybrid,offload` would die on moe_cache_size=0. + if not any(k in engine_kwargs + for k in ("moe_cache_size", "moe_cache_rate", "moe_cache_auto")): + engine_kwargs = {**engine_kwargs, "moe_cache_auto": True} + llm = LLM(model_path, dtype=torch.bfloat16, **engine_kwargs) # `ignore_eos` so every run generates exactly `tokens` -- otherwise an early stop # silently shortens the measurement and inflates the rate. From c95a608b352d1ee07e9ccda1f846559baaad3018 Mon Sep 17 00:00:00 2001 From: "Gabriel A. Devenyi" Date: Sat, 22 Aug 2026 14:36:12 -0400 Subject: [PATCH 3/3] feat(bench decode): --concurrency, so batch-dependent costs are measurable The command only ever generated one stream, which is the single-stream latency case rather than the serving case. Anything whose cost depends on how many tokens a decode step carries was invisible: expert dedup on the CPU MoE path, CUDA-graph batch selection, the scheduler itself. Having just added expert dedup and then being unable to measure it with this tool is the argument. `--concurrency N` generates N streams at once and reports aggregate tokens/s. The offline LLM already takes a list of prompts and decodes them together, so no server or client fan-out is involved. Two details that would otherwise make it measure the wrong thing: * `max_running_req` and `cuda_graph_max_bs` default up to N. Without room for the streams the scheduler serializes them, the decode batch never grows, and the run silently reports concurrency 1. * The prompts are made distinct. The radix cache shares the KV of identical prefixes, so N copies of one prompt would measure one stream plus N-1 cache hits. On DeepSeek-V4-Flash it immediately answers a question the single-stream mode could not: offload beats hybrid by 3.9% at one stream and by 15.7% at eight, so the gap widens with load rather than closing. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GpGe2fQ5pDShGrnSuksfun --- python/freetoken/moe/bench_decode.py | 41 ++++++++++++++++++++++------ 1 file changed, 32 insertions(+), 9 deletions(-) diff --git a/python/freetoken/moe/bench_decode.py b/python/freetoken/moe/bench_decode.py index f8ed71e1d..27a8c18be 100644 --- a/python/freetoken/moe/bench_decode.py +++ b/python/freetoken/moe/bench_decode.py @@ -59,8 +59,15 @@ def _coerce(v: str): def measure_decode_tps(model_path: str, engine_kwargs: dict, prompt: str, - tokens: int, samples: int) -> dict: - """Decode tokens/s for one engine configuration, median over ``samples``.""" + tokens: int, samples: int, concurrency: int = 1) -> dict: + """Decode tokens/s for one engine configuration, median over ``samples``. + + ``concurrency`` generates that many streams at once, so decode steps carry that + many tokens. Anything whose cost depends on the decode batch -- expert dedup on + the CPU MoE path, CUDA-graph batch selection, the scheduler itself -- is invisible + at concurrency 1, which is the single-stream latency case, not the serving case. + Reported tokens/s is aggregate across the streams. + """ import torch from freetoken.core import SamplingParams @@ -72,6 +79,11 @@ def measure_decode_tps(model_path: str, engine_kwargs: dict, prompt: str, if not any(k in engine_kwargs for k in ("moe_cache_size", "moe_cache_rate", "moe_cache_auto")): engine_kwargs = {**engine_kwargs, "moe_cache_auto": True} + if concurrency > 1: + # Without room for the streams the scheduler just serializes them and the + # decode batch never grows, which silently measures concurrency 1. + engine_kwargs.setdefault("max_running_req", concurrency) + engine_kwargs.setdefault("cuda_graph_max_bs", concurrency) llm = LLM(model_path, dtype=torch.bfloat16, **engine_kwargs) # `ignore_eos` so every run generates exactly `tokens` -- otherwise an early stop @@ -79,20 +91,23 @@ def measure_decode_tps(model_path: str, engine_kwargs: dict, prompt: str, one = SamplingParams(max_tokens=1, temperature=0.0, ignore_eos=True) many = SamplingParams(max_tokens=tokens, temperature=0.0, ignore_eos=True) - llm.generate([prompt], SamplingParams(max_tokens=8, temperature=0.0, ignore_eos=True)) + # Distinct prompts: the radix cache would share the KV of identical prefixes, so + # N copies of one prompt measures one stream plus N-1 cache hits. + prompts = [f"{prompt} (variant {i})" for i in range(concurrency)] + llm.generate(prompts, SamplingParams(max_tokens=8, temperature=0.0, ignore_eos=True)) rates = [] for _ in range(samples): t0 = time.perf_counter() - llm.generate([prompt], one) + llm.generate(prompts, one) t_prefill = time.perf_counter() - t0 t0 = time.perf_counter() - llm.generate([prompt], many) + llm.generate(prompts, many) t_total = time.perf_counter() - t0 decode_s = t_total - t_prefill if decode_s <= 0: continue - rates.append((tokens - 1) / decode_s) + rates.append(concurrency * (tokens - 1) / decode_s) if not rates: raise RuntimeError("no usable timing samples") return { @@ -107,7 +122,8 @@ def _run_worker(argv: list[str]) -> int: spec = json.loads(argv[0]) try: out = measure_decode_tps(spec["model"], spec["kwargs"], spec["prompt"], - spec["tokens"], spec["samples"]) + spec["tokens"], spec["samples"], + spec.get("concurrency", 1)) except Exception as e: # noqa: BLE001 - reported to the parent, not swallowed print(json.dumps({"error": f"{type(e).__name__}: {e}"}), flush=True) return 1 @@ -162,6 +178,10 @@ def main(argv: list[str] | None = None, prog: str = "ft bench decode") -> int: p.add_argument("--samples", type=int, default=3, help="timed generations per load (default 3)") p.add_argument("--tokens", type=int, default=128, help="tokens per generation") + p.add_argument("--concurrency", type=int, default=1, metavar="N", + help="generate N streams at once, so decode steps carry N tokens " + "(default 1). Reported tokens/s is aggregate. Anything whose " + "cost depends on the decode batch is invisible at 1") p.add_argument("--prompt", default=DEFAULT_PROMPT) p.add_argument("-o", "--out", default=None, help="write results as JSON") p.add_argument("-q", "--quiet", action="store_true") @@ -173,13 +193,15 @@ def main(argv: list[str] | None = None, prog: str = "ft bench decode") -> int: variants = _variants(ns.compare, ns.set) results: dict[str, list[float]] = {name: [] for name, _ in variants} print(f" {len(variants)} variant(s) x {ns.cycles} cycle(s), " - f"{ns.samples} timed generations of {ns.tokens} tokens each") + f"{ns.samples} timed generations of {ns.tokens} tokens each" + + (f", {ns.concurrency} streams at once" if ns.concurrency > 1 else "")) print(" each measurement reloads the model in a fresh process\n") for cycle in range(1, ns.cycles + 1): for name, kwargs in variants: spec = {"model": ns.model, "kwargs": kwargs, "prompt": ns.prompt, - "tokens": ns.tokens, "samples": ns.samples} + "tokens": ns.tokens, "samples": ns.samples, + "concurrency": ns.concurrency} t0 = time.perf_counter() out = _measure_in_subprocess(spec, ns.quiet) dt = time.perf_counter() - t0 @@ -208,6 +230,7 @@ def main(argv: list[str] | None = None, prog: str = "ft bench decode") -> int: if ns.out: with open(ns.out, "w") as f: json.dump({"model": ns.model, "tokens": ns.tokens, + "concurrency": ns.concurrency, "results": {n: v for n, v in results.items()}}, f, indent=2) f.write("\n") print(f"\n saved: {ns.out}")