diff --git a/.github/workflows/research-qwen38-benchmark.yml b/.github/workflows/research-qwen38-benchmark.yml new file mode 100644 index 000000000..a1df99f8c --- /dev/null +++ b/.github/workflows/research-qwen38-benchmark.yml @@ -0,0 +1,35 @@ +name: Qwen3.8 research benchmark harness + +on: + push: + branches: + - research/qwen38-3090-benchmark-ready + workflow_dispatch: + +permissions: + contents: read + +jobs: + benchmark-harness: + runs-on: windows-latest + timeout-minutes: 10 + steps: + - name: Check out source + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install test runner + run: python -m pip install pytest + + - name: Compile benchmark and tests + run: >- + python -m py_compile + benchmarks/bench_decode_moe.py + tests/benchmarks/test_bench_decode_moe.py + + - name: Run focused tests + run: python -m pytest -q tests/benchmarks/test_bench_decode_moe.py diff --git a/benchmarks/bench_decode_moe.py b/benchmarks/bench_decode_moe.py index 566217927..f5c26e2e9 100644 --- a/benchmarks/bench_decode_moe.py +++ b/benchmarks/bench_decode_moe.py @@ -37,8 +37,13 @@ CUDA_VISIBLE_DEVICES=0 PYTHONPATH=python python benchmarks/bench_decode_moe.py \ --model /path/to/model -Run (all three backends, one server per backend): - ... --model /path/to/model --backend offload,cpu,hybrid --json out.json +Exact Windows Qwen3.8 research run: + python benchmarks/bench_decode_moe.py \ + --model C:\\Models\\Qwen3.8-Flash-Next-NVFP4-FTW \ + --backend offload --nvfp4-backend triton --ple-backend mmap \ + --expert-load serial --cache 2048 --max-seq-len 131072 \ + --kv-reserve-tokens 131072 --decode 256 --greedy \ + --temp-dir .bench-temp --json qwen38-128k.jsonl """ from __future__ import annotations @@ -58,78 +63,111 @@ import urllib.request from pathlib import Path -# Applied for every field the checkpoint's generation_config.json does not specify. FALLBACK_SAMPLING = {"temperature": 1.0, "top_p": 0.95, "top_k": 64} - -# AIME-25 problems, pulled from the Hub into the usual HF cache on first run. AIME_REPO = "math-ai/aime25" AIME_FILE = "test.jsonl" -# Reasoning models need the answer format spelled out; the boxed answer is also what makes -# a run spot-checkable by eye. BOXED_INSTRUCTION = ( "Please reason step by step, and put your final answer within \\boxed{}." ) def parse_args(argv: list[str] | None = None) -> argparse.Namespace: - p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - p.add_argument("--model", required=True, help="checkpoint dir (or .ftw)") - p.add_argument( + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--model", required=True, help="checkpoint dir (or .ftw)") + parser.add_argument( "--backend", default="offload", help="comma list of offload|cpu|hybrid; one server per backend", ) - p.add_argument( + parser.add_argument( "--aime", default=os.environ.get("FREETOKEN_AIME25_JSONL"), help=f"local jsonl instead of downloading {AIME_REPO}; default $FREETOKEN_AIME25_JSONL", ) - p.add_argument("--problem", type=int, default=0, help="0-based AIME problem index") - p.add_argument("--decode", type=int, default=256, help="decode tokens to measure (D)") - p.add_argument( + parser.add_argument("--problem", type=int, default=0, help="0-based AIME problem index") + parser.add_argument("--decode", type=int, default=256, help="decode tokens to measure") + parser.add_argument( "--cache", type=int, default=0, help="GPU expert cache slots; 0 = auto-size from free VRAM", ) - p.add_argument("--cache-rate", type=float, default=None, help="cache slots as a fraction of L*E") - p.add_argument( + parser.add_argument( + "--cache-rate", type=float, default=None, help="cache slots as a fraction of L*E" + ) + parser.add_argument( "--hybrid-fetch", type=int, default=-1, - help="hybrid: max PCIe fetches/layer; -1 = auto (benched pcie/cpu bandwidth fraction)", + help="hybrid: maximum PCIe fetches per layer; -1 = auto", ) - p.add_argument("--mem-ratio", type=float, default=0.9, help="target VRAM utilization") - p.add_argument("--gpu", default=None, - help="GPU for the serve: a UUID or nvidia-smi index (as ft serve --gpu)") - p.add_argument("--no-graph", action="store_true", help="eager decode instead of CUDA graph") - p.add_argument( + parser.add_argument("--mem-ratio", type=float, default=0.9, help="target VRAM utilization") + parser.add_argument("--gpu", default=None, help="GPU UUID or nvidia-smi index") + parser.add_argument("--no-graph", action="store_true", help="use eager decode") + parser.add_argument( "--greedy", action="store_true", - help="force temperature 0 (ignore the checkpoint's sampling) so ids are comparable", + help="force temperature 0 so output is comparable", ) - p.add_argument( + parser.add_argument( "--server-timeout", type=float, default=1800, - help="seconds to wait for the spawned server to become ready", + help="seconds to wait for the spawned server", + ) + parser.add_argument( + "--max-seq-len", + type=int, + default=0, + help="physical sequence capacity; 0 keeps the historical 8192+decode value", + ) + parser.add_argument( + "--kv-reserve-tokens", + type=int, + default=0, + help="KV tokens reserved before automatic expert-cache sizing; 0 omits the flag", + ) + parser.add_argument( + "--ple-backend", + choices=("pinned", "mmap"), + default="pinned", + help="Qwen3.8 PLE table storage", + ) + parser.add_argument( + "--expert-load", + choices=("auto", "serial", "parallel"), + default="auto", + help="host expert-bank load strategy", ) - p.add_argument("--json", dest="json_out", default=None, help="append the result rows here") - return p.parse_args(argv) + parser.add_argument("--attention-backend", default="auto") + parser.add_argument( + "--nvfp4-backend", + choices=("auto", "triton", "marlin", "flashinfer"), + default="auto", + ) + parser.add_argument( + "--temp-dir", + default=os.environ.get("FREETOKEN_BENCH_TEMP"), + help="writable directory for logs and child-process temporary files", + ) + parser.add_argument("--json", dest="json_out", default=None, help="append result rows here") + return parser.parse_args(argv) def load_problem(path: str | None, index: int) -> tuple[str, str]: - """One AIME-25 (problem, answer). Downloads the dataset unless ``path`` overrides it. - - Accepts both the Hub schema (``problem``) and the pre-formatted jsonl some local copies - use (``prompt``, answer instruction already appended).""" + """Load one AIME-25 problem and its expected answer.""" if not path: from huggingface_hub import hf_hub_download try: path = hf_hub_download(AIME_REPO, AIME_FILE, repo_type="dataset") - except Exception as e: # offline, rate-limited, repo moved - sys.exit(f"could not fetch {AIME_REPO}/{AIME_FILE} ({e}); pass --aime ") + except Exception as error: + sys.exit( + f"could not fetch {AIME_REPO}/{AIME_FILE} ({error}); " + "pass --aime " + ) rows = [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()] if not 0 <= index < len(rows): sys.exit(f"--problem {index} out of range ({len(rows)} problems available)") @@ -141,74 +179,127 @@ def load_problem(path: str | None, index: int) -> tuple[str, str]: def resolve_sampling(model_path: str, greedy: bool) -> tuple[dict, str]: - """Checkpoint-recommended sampling with per-field fallback; returns (params, source). - - Resolved client-side and sent explicitly: the server fills unspecified fields with - its framework defaults (temperature 0 / no filtering), not with these fallbacks.""" + """Return explicit sampling parameters and their source.""" if greedy: return {"temperature": 0.0, "top_p": 1.0, "top_k": -1}, "greedy (--greedy)" recommended: dict = {} - cfg = Path(model_path) / "generation_config.json" - if cfg.is_file(): - raw = json.loads(cfg.read_text()) - recommended = {k: raw[k] for k in FALLBACK_SAMPLING if raw.get(k) is not None} + config_path = Path(model_path) / "generation_config.json" + if config_path.is_file(): + raw = json.loads(config_path.read_text()) + recommended = {key: raw[key] for key in FALLBACK_SAMPLING if raw.get(key) is not None} if raw.get("do_sample") is False or recommended.get("temperature") == 0.0: return {"temperature": 0.0, "top_p": 1.0, "top_k": -1}, "checkpoint (greedy)" params = {**FALLBACK_SAMPLING, **recommended} if params["top_k"] == 0: - params["top_k"] = -1 # HF spells "no top-k filtering" as 0; the API as -1 - taken = sorted(recommended) - source = f"checkpoint{taken} + fallback" if taken else "fallback (no generation_config)" + params["top_k"] = -1 + selected = sorted(recommended) + source = f"checkpoint{selected} + fallback" if selected else "fallback (no generation_config)" return params, source def get_json(url: str, timeout: float = 10) -> dict: - with urllib.request.urlopen(url, timeout=timeout) as resp: - return json.load(resp) + with urllib.request.urlopen(url, timeout=timeout) as response: + return json.load(response) def free_port() -> int: - with socket.socket() as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] + with socket.socket() as server_socket: + server_socket.bind(("127.0.0.1", 0)) + return server_socket.getsockname()[1] + + +def resolved_max_seq_len(args: argparse.Namespace) -> int: + if args.max_seq_len > 0: + return args.max_seq_len + return 8192 + args.decode + + +def resolve_temp_dir(raw_path: str | None) -> Path | None: + if raw_path is None: + return None + temp_dir = Path(raw_path).resolve() + temp_dir.mkdir(parents=True, exist_ok=True) + return temp_dir + + +def server_environment(temp_dir: Path | None) -> dict[str, str]: + environment = os.environ.copy() + source_python = Path(__file__).resolve().parents[1] / "python" + current_python_path = environment.get("PYTHONPATH") + python_paths = [str(source_python)] + if current_python_path: + python_paths.append(current_python_path) + environment["PYTHONPATH"] = os.pathsep.join(python_paths) + environment["PYTHONUNBUFFERED"] = "1" + if temp_dir is not None: + environment["TEMP"] = str(temp_dir) + environment["TMP"] = str(temp_dir) + return environment def serve_cmd(args: argparse.Namespace, backend: str, port: int) -> list[str]: - cmd = [ - sys.executable, "-m", "freetoken.cli", "serve", - "--model", args.model, - "--host", "127.0.0.1", "--port", str(port), - "--moe-backend", backend, - "--max-running-requests", "1", - "--max-seq-len-override", str(8192 + args.decode), - "--memory-ratio", str(args.mem_ratio), - "--cuda-graph-max-bs", "0" if args.no_graph else "1", - "--moe-hybrid-max-fetch", str(args.hybrid_fetch), + command = [ + sys.executable, + "-m", + "freetoken.cli", + "serve", + "--model", + args.model, + "--host", + "127.0.0.1", + "--port", + str(port), + "--moe-backend", + backend, + "--max-running-requests", + "1", + "--max-seq-len-override", + str(resolved_max_seq_len(args)), + "--memory-ratio", + str(args.mem_ratio), + "--cuda-graph-max-bs", + "0" if args.no_graph else "1", + "--moe-hybrid-max-fetch", + str(args.hybrid_fetch), + "--attention-backend", + args.attention_backend, + "--nvfp4-backend", + args.nvfp4_backend, + "--expert-load", + args.expert_load, + "--ple-backend", + args.ple_backend, ] + if args.kv_reserve_tokens > 0: + command += ["--kv-reserve-tokens", str(args.kv_reserve_tokens)] if args.gpu: - cmd += ["--gpu", args.gpu] + command += ["--gpu", args.gpu] if args.cache > 0: - cmd += ["--moe-cache-size", str(args.cache)] + command += ["--moe-cache-size", str(args.cache)] elif args.cache_rate is not None: - cmd += ["--moe-cache-rate", str(args.cache_rate)] + command += ["--moe-cache-rate", str(args.cache_rate)] else: - cmd.append("--moe-cache-auto") - return cmd + command.append("--moe-cache-auto") + return command -def die_with_log(msg: str, log_path: str) -> None: - tail = "".join(Path(log_path).read_text().splitlines(keepends=True)[-30:]) - sys.exit(f"[bench] {msg}\n[bench] server log tail ({log_path}):\n{tail}") +def die_with_log(message: str, log_path: str) -> None: + tail = "".join( + Path(log_path).read_text(errors="replace").splitlines(keepends=True)[-30:] + ) + sys.exit(f"[bench] {message}\n[bench] server log tail ({log_path}):\n{tail}") -def wait_ready(origin: str, proc: subprocess.Popen, log_path: str, timeout: float) -> None: +def wait_ready(origin: str, process: subprocess.Popen, log_path: str, timeout: float) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: - if proc.poll() is not None: - die_with_log(f"server exited with code {proc.returncode} during startup", log_path) + if process.poll() is not None: + die_with_log( + f"server exited with code {process.returncode} during startup", log_path + ) try: health = get_json(f"{origin}/health", timeout=5) - except (OSError, ValueError): # not bound yet / reset / partial response + except (OSError, ValueError): time.sleep(1.0) continue if health.get("status") == "error": @@ -219,39 +310,85 @@ def wait_ready(origin: str, proc: subprocess.Popen, log_path: str, timeout: floa die_with_log(f"server not ready after {timeout:.0f}s", log_path) -def pump_output(src, log_f) -> None: - """Mirror the server's output to our terminal while keeping the log file complete. - - Raw byte chunks (read1, not line-buffered) so \\r progress bars render live.""" - for chunk in iter(lambda: src.read1(65536), b""): - log_f.write(chunk) - log_f.flush() +def pump_output(source, log_file) -> None: + """Mirror server bytes to the terminal and the persistent log.""" + for chunk in iter(lambda: source.read1(65536), b""): + log_file.write(chunk) + log_file.flush() sys.stdout.buffer.write(chunk) sys.stdout.flush() -def stop_server(proc: subprocess.Popen) -> None: - """SIGTERM the whole session (frontend + scheduler/tokenizer workers), escalate. +def start_server(command: list[str], environment: dict[str, str]) -> subprocess.Popen: + common = { + "stdout": subprocess.PIPE, + "stderr": subprocess.STDOUT, + "env": environment, + } + if os.name == "nt": + return subprocess.Popen( + command, + creationflags=subprocess.CREATE_NEW_PROCESS_GROUP, + **common, + ) + return subprocess.Popen(command, start_new_session=True, **common) + - Best-effort by design: it runs in ``finally`` and must not mask the real error. - killpg runs even when the frontend already exited -- a crashed frontend leaves live - non-daemon workers in the group, and they hold the GPU.""" - for sig, wait_s in ((signal.SIGTERM, 90), (signal.SIGKILL, 30)): - try: - os.killpg(proc.pid, sig) - except ProcessLookupError: # whole group already gone - pass +def wait_for_exit(process: subprocess.Popen, timeout: float) -> bool: + try: + process.wait(timeout=timeout) + return True + except subprocess.TimeoutExpired: + return False + + +def stop_windows_process_tree(process: subprocess.Popen) -> None: + try: + process.send_signal(signal.CTRL_BREAK_EVENT) + except (OSError, ValueError): + pass + if wait_for_exit(process, 20): + return + subprocess.run( + ["taskkill", "/PID", str(process.pid), "/T", "/F"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + wait_for_exit(process, 30) + + +def stop_posix_process_group(process: subprocess.Popen) -> None: + for process_signal, wait_seconds in ((signal.SIGTERM, 90), (signal.SIGKILL, 30)): try: - proc.wait(timeout=wait_s) - break - except subprocess.TimeoutExpired: - continue - time.sleep(3) # let the driver reclaim VRAM before the next backend's server + os.killpg(process.pid, process_signal) + except ProcessLookupError: + return + if wait_for_exit(process, wait_seconds): + return -def stream_generate(origin: str, model_id: str, problem: str, sampling: dict, - args: argparse.Namespace) -> dict: - """One streamed chat completion; returns per-token arrival stamps, text, and usage.""" +def stop_server(process: subprocess.Popen) -> None: + """Stop only the process tree that this benchmark started.""" + if process.poll() is None: + if os.name == "nt": + stop_windows_process_tree(process) + else: + stop_posix_process_group(process) + if process.poll() is None: + process.kill() + wait_for_exit(process, 10) + time.sleep(3) + + +def stream_generate( + origin: str, + model_id: str, + problem: str, + sampling: dict, + args: argparse.Namespace, +) -> dict: + """Run one streamed completion and return arrival times, text, and usage.""" body = { "model": model_id, "messages": [{"role": "user", "content": problem}], @@ -262,7 +399,7 @@ def stream_generate(origin: str, model_id: str, problem: str, sampling: dict, "chat_template_kwargs": {"enable_thinking": True}, **sampling, } - req = urllib.request.Request( + request = urllib.request.Request( f"{origin}/v1/chat/completions", data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, @@ -270,20 +407,17 @@ def stream_generate(origin: str, model_id: str, problem: str, sampling: dict, stamps: list[float] = [] pieces: list[str] = [] usage: dict | None = None - t0 = time.perf_counter() + start_time = time.perf_counter() try: - resp = urllib.request.urlopen(req, timeout=1800) - except urllib.error.HTTPError as e: - sys.exit(f"[bench] request failed: HTTP {e.code}: {e.read()[:500]!r}") - # Iterate the SSE stream line by line as bytes; json.loads decodes UTF-8 itself. - # (A text-mode reader keyed off the content-type would decode latin-1: the server - # sends ensure_ascii=False JSON with no charset on text/event-stream.) - with resp: - for raw in resp: + response = urllib.request.urlopen(request, timeout=1800) + except urllib.error.HTTPError as error: + sys.exit(f"[bench] request failed: HTTP {error.code}: {error.read()[:500]!r}") + with response: + for raw in response: line = raw.strip() if not line or not line.startswith(b"data:"): - continue # blank separators between events - payload = line[len(b"data:"):].strip() + continue + payload = line[len(b"data:") :].strip() if payload == b"[DONE]": break now = time.perf_counter() @@ -297,56 +431,68 @@ def stream_generate(origin: str, model_id: str, problem: str, sampling: dict, stamps.append(now) pieces.append(text) if usage is None: - sys.exit("[bench] stream ended without a usage chunk; is this a FreeToken server?") - return {"t0": t0, "stamps": stamps, "text": "".join(pieces), "usage": usage} + sys.exit("[bench] stream ended without a usage chunk") + return {"t0": start_time, "stamps": stamps, "text": "".join(pieces), "usage": usage} def run_one(args: argparse.Namespace, backend: str) -> dict: problem, answer = load_problem(args.aime, args.problem) - sampling, sampling_src = resolve_sampling(args.model, args.greedy) + sampling, sampling_source = resolve_sampling(args.model, args.greedy) port = free_port() origin = f"http://127.0.0.1:{port}" - fd, log_path = tempfile.mkstemp(prefix=f"bench-serve-{backend}-", suffix=".log") - cmd = serve_cmd(args, backend, port) + temp_dir = resolve_temp_dir(args.temp_dir) + fd, log_path = tempfile.mkstemp( + prefix=f"bench-serve-{backend}-", + suffix=".log", + dir=str(temp_dir) if temp_dir is not None else None, + ) + command = serve_cmd(args, backend, port) print( f"[bench] model={args.model}\n" f"[bench] backend={backend} cache={args.cache or args.cache_rate or 'auto'} " - f"mem_ratio={args.mem_ratio} decode={args.decode} graph={not args.no_graph}\n" - f"[bench] sampling={sampling} <- {sampling_src}\n" + f"max_seq_len={resolved_max_seq_len(args)} kv_reserve={args.kv_reserve_tokens or 'default'}\n" + f"[bench] ple={args.ple_backend} expert_load={args.expert_load} " + f"attention={args.attention_backend} nvfp4={args.nvfp4_backend}\n" + f"[bench] mem_ratio={args.mem_ratio} decode={args.decode} graph={not args.no_graph}\n" + f"[bench] sampling={sampling} <- {sampling_source}\n" f"[bench] server log: {log_path}", flush=True, ) - with os.fdopen(fd, "wb") as log_f: - proc = subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=True + with os.fdopen(fd, "wb") as log_file: + process = start_server(command, server_environment(temp_dir)) + if process.stdout is None: + raise RuntimeError("server stdout pipe was not created") + pump = threading.Thread( + target=pump_output, args=(process.stdout, log_file), daemon=True ) - pump = threading.Thread(target=pump_output, args=(proc.stdout, log_f), daemon=True) pump.start() try: - wait_ready(origin, proc, log_path, args.server_timeout) + wait_ready(origin, process, log_path, args.server_timeout) model_id = get_json(f"{origin}/v1/models")["data"][0]["id"] print(f"[bench] model_id={model_id}", flush=True) print(f"[bench] AIME25 #{args.problem} (answer {answer})", flush=True) - - # Warm the expert cache to a steady-state decode working set. stream_generate(origin, model_id, problem, sampling, args) - r = stream_generate(origin, model_id, problem, sampling, args) + result = stream_generate(origin, model_id, problem, sampling, args) stats = get_json(f"{origin}/v1/stats") finally: - stop_server(proc) + stop_server(process) pump.join(timeout=10) - stamps, usage = r["stamps"], r["usage"] + stamps = result["stamps"] + usage = result["usage"] if len(stamps) < 2: - sys.exit(f"[bench] need >=2 token events to measure decode, got {len(stamps)}") + sys.exit(f"[bench] need at least 2 token events, got {len(stamps)}") completion = usage["completion_tokens"] if completion != args.decode: - print(f"[bench] WARNING: completion_tokens={completion} != --decode {args.decode}", flush=True) + print( + f"[bench] WARNING: completion_tokens={completion} != --decode {args.decode}", + flush=True, + ) steps = completion - 1 decode_time = stamps[-1] - stamps[0] - gaps = sorted((b - a) * 1e3 for a, b in zip(stamps, stamps[1:])) + gaps = sorted((end - start) * 1e3 for start, end in zip(stamps, stamps[1:])) row = { "model": args.model, "backend": backend, @@ -357,50 +503,59 @@ def run_one(args: argparse.Namespace, backend: str) -> dict: "ms_per_token": decode_time / steps * 1e3 if steps > 0 else 0.0, "event_ms_p50": gaps[len(gaps) // 2], "event_ms_p99": gaps[min(len(gaps) - 1, int(len(gaps) * 0.99))], - "ttft_ms": (stamps[0] - r["t0"]) * 1e3, + "ttft_ms": (stamps[0] - result["t0"]) * 1e3, "events": len(stamps), "completion_tokens": completion, "vram_gib": stats.get("vram_bytes", 0) / 2**30, "sampling": sampling, - "output_sha1": hashlib.sha1(r["text"].encode()).hexdigest()[:12], + "output_sha1": hashlib.sha1(result["text"].encode()).hexdigest()[:12], "server_log": log_path, + "max_seq_len": resolved_max_seq_len(args), + "kv_reserve_tokens": args.kv_reserve_tokens, + "moe_cache_size": args.cache, + "ple_backend": args.ple_backend, + "expert_load": args.expert_load, + "attention_backend": args.attention_backend, + "nvfp4_backend": args.nvfp4_backend, } print(f"\n==== decode bs=1 [{backend}] via /v1/chat/completions ====", flush=True) print(f" decode throughput : {row['decode_tok_s']:8.2f} tok/s ({row['ms_per_token']:.3f} ms/token)") print(f" TTFT (warm) : {row['ttft_ms']:8.1f} ms (prompt {row['prompt_tokens']} tok)") - print(f" decode measured : {steps} steps in {decode_time:.3f} s " - f"(event p50 {row['event_ms_p50']:.3f} / p99 {row['event_ms_p99']:.3f} ms, " - f"{len(stamps)} events)") + print( + f" decode measured : {steps} steps in {decode_time:.3f} s " + f"(event p50 {row['event_ms_p50']:.3f} / p99 {row['event_ms_p99']:.3f} ms, " + f"{len(stamps)} events)" + ) print(f" vram (server) : {row['vram_gib']:8.2f} GiB") sha_note = "greedy" if args.greedy else "sampled, per-server deterministic" - print(f" output sha1 : {row['output_sha1']} ({sha_note}; compare across backends)") - print(f" output sample : {r['text'][:240]!r}") + print(f" output sha1 : {row['output_sha1']} ({sha_note})") + print(f" output sample : {result['text'][:240]!r}") return row def main(argv: list[str] | None = None) -> int: args = parse_args(argv) - backends = [b.strip() for b in args.backend.split(",") if b.strip()] - unknown = [b for b in backends if b not in ("offload", "cpu", "hybrid")] - if unknown: - sys.exit(f"unknown backend(s): {unknown}") + backends = [backend.strip() for backend in args.backend.split(",") if backend.strip()] + unsupported = [ + backend for backend in backends if backend not in ("offload", "cpu", "hybrid") + ] + if unsupported: + sys.exit(f"unsupported backend(s): {unsupported}") - failed = [] + failed: list[str] = [] for backend in backends: try: row = run_one(args, backend) - # SystemExit inherits BaseException, not Exception, so name both: a mid-decode - # connection drop (server crash) must not abort the remaining backends either. - except (SystemExit, Exception) as e: + except (SystemExit, Exception) as error: if len(backends) == 1: raise - print(f"\n[bench] backend {backend} failed: {e!r}", flush=True) + print(f"\n[bench] backend {backend} failed: {error!r}", flush=True) failed.append(backend) continue if args.json_out: - with open(args.json_out, "a") as f: - f.write(json.dumps(row) + "\n") + with open(args.json_out, "a", encoding="utf-8") as output_file: + output_file.write(json.dumps(row) + "\n") if failed: print(f"\n[bench] backends that failed: {failed}", flush=True) return 1 diff --git a/tests/benchmarks/test_bench_decode_moe.py b/tests/benchmarks/test_bench_decode_moe.py new file mode 100644 index 000000000..063ec7eb3 --- /dev/null +++ b/tests/benchmarks/test_bench_decode_moe.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import importlib.util +import os +import sys +from pathlib import Path +from types import ModuleType + + +def load_benchmark_module() -> ModuleType: + script_path = Path(__file__).resolve().parents[2] / "benchmarks" / "bench_decode_moe.py" + spec = importlib.util.spec_from_file_location("bench_decode_moe_test_module", script_path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def command_value(command: list[str], option: str) -> str: + return command[command.index(option) + 1] + + +def test_exact_qwen38_geometry_reaches_server_command(tmp_path: Path) -> None: + benchmark = load_benchmark_module() + args = benchmark.parse_args( + [ + "--model", + "C:/Models/Qwen3.8-Flash-Next-NVFP4-FTW", + "--backend", + "offload", + "--nvfp4-backend", + "triton", + "--ple-backend", + "mmap", + "--expert-load", + "serial", + "--attention-backend", + "qsa_sparse", + "--cache", + "2048", + "--max-seq-len", + "131072", + "--kv-reserve-tokens", + "131072", + "--decode", + "256", + "--temp-dir", + str(tmp_path), + ] + ) + + command = benchmark.serve_cmd(args, "offload", 8123) + + assert command_value(command, "--max-seq-len-override") == "131072" + assert command_value(command, "--kv-reserve-tokens") == "131072" + assert command_value(command, "--moe-cache-size") == "2048" + assert command_value(command, "--ple-backend") == "mmap" + assert command_value(command, "--expert-load") == "serial" + assert command_value(command, "--attention-backend") == "qsa_sparse" + assert command_value(command, "--nvfp4-backend") == "triton" + assert "--moe-cache-auto" not in command + + +def test_default_geometry_and_auto_cache_are_preserved() -> None: + benchmark = load_benchmark_module() + args = benchmark.parse_args(["--model", "model", "--decode", "256"]) + + command = benchmark.serve_cmd(args, "offload", 8123) + + assert benchmark.resolved_max_seq_len(args) == 8448 + assert command_value(command, "--max-seq-len-override") == "8448" + assert "--kv-reserve-tokens" not in command + assert "--moe-cache-auto" in command + + +def test_server_environment_pins_source_and_temp_paths(tmp_path: Path) -> None: + benchmark = load_benchmark_module() + environment = benchmark.server_environment(tmp_path) + expected_source = str(Path(benchmark.__file__).resolve().parents[1] / "python") + + assert environment["PYTHONPATH"].split(os.pathsep)[0] == expected_source + assert environment["PYTHONUNBUFFERED"] == "1" + assert environment["TEMP"] == str(tmp_path) + assert environment["TMP"] == str(tmp_path) + + +def test_resolve_temp_dir_creates_requested_directory(tmp_path: Path) -> None: + benchmark = load_benchmark_module() + requested = tmp_path / "nested" / "bench" + + resolved = benchmark.resolve_temp_dir(str(requested)) + + assert resolved == requested.resolve() + assert requested.is_dir() + + +def test_start_server_captures_output() -> None: + benchmark = load_benchmark_module() + command = [sys.executable, "-c", "print('BENCH_CHILD_OK')"] + + process = benchmark.start_server(command, os.environ.copy()) + output, _ = process.communicate(timeout=10) + + assert process.returncode == 0 + assert output.strip() == b"BENCH_CHILD_OK"