#!/usr/bin/env python3 """Run repeatable CPU inference measurements and save full prompt/answer data.""" from __future__ import annotations import json import os import signal import subprocess import sys import time from datetime import datetime, timezone import demo PROMPTS = [ "請用繁體中文一句話說明 SSD 和 RAM 的差異。", "請列出 1 到 5 的平方，格式為「數字:平方」，不要加其他說明。", "Translate to English: 今天天氣很好，我想去公園散步。", "請用 Python 寫一個計算階乘的函式，程式碼不超過 4 行。", ] REPEATS = int(os.environ.get("BENCH_REPEATS", "2")) MAX_TOKENS = int(os.environ.get("BENCH_MAX_TOKENS", "128")) BENCH_LOG = demo.VALIDATE_DIR / "benchmark-server.log" RESULT_PATH = demo.VALIDATE_DIR / "performance.json" def cpu_name() -> str: try: for line in open("/proc/cpuinfo", encoding="utf-8"): if line.lower().startswith("model name"): return line.split(":", 1)[1].strip() except OSError: pass return "unknown" def main() -> int: if REPEATS < 1 or MAX_TOKENS < 1: raise SystemExit("BENCH_REPEATS 與 BENCH_MAX_TOKENS 必須大於 0") if not demo.BIN.exists() or not demo.MODEL_PATH.exists(): raise SystemExit("請先執行 python3 demo.py 完成安裝與模型下載") demo.VALIDATE_DIR.mkdir(parents=True, exist_ok=True) kv_path = demo.MODEL_DIR / "kv-ssd.cache" env = os.environ.copy() env.update({ "LLAMA_MMAP_PREFETCH": "0", "LLAMA_MMAP_RELEASE": "1", "LLAMA_MMAP_RELEASE_INTERVAL": str(demo.MMAP_RELEASE_INTERVAL_MS), "LLAMA_KV_SSD": "1", "LLAMA_KV_SSD_MIN_MIB": "0", "LLAMA_KV_SSD_PATH": str(kv_path), }) command = [ str(demo.BIN), "-m", str(demo.MODEL_PATH), "--alias", "qwen3-0.6b-q8_0", "--jinja", "--load-mode", "mmap", "--fit", "off", "--no-repack", "--no-warmup", "-t", str(demo.THREADS), "-tb", str(demo.THREADS), "-c", str(demo.CONTEXT), "-b", "128", "-ub", "64", "--cache-type-k", "q8_0", "--cache-type-v", "q8_0", "-np", "1", "--host", demo.HOST, "--port", str(demo.PORT), ] records: list[dict[str, object]] = [] with BENCH_LOG.open("w", encoding="utf-8") as log: process = subprocess.Popen(command, cwd=demo.ROOT, env=env, stdout=log, stderr=subprocess.STDOUT) monitor = demo.ProcessMonitor(process.pid) monitor.start() pagecache = demo.PageCacheReclaimer([demo.MODEL_PATH, kv_path]) pagecache.start() try: demo.wait_ready(process) system_swap_before = demo.system_swap() warmup_payload = demo.make_prompt("請只回答數字：1 加 1 等於多少？") warmup_payload["max_tokens"] = 32 demo.api_request("/v1/chat/completions", warmup_payload) for repeat in range(1, REPEATS + 1): for prompt in PROMPTS: payload = demo.make_prompt(prompt) payload["max_tokens"] = MAX_TOKENS started = time.perf_counter() response = demo.api_request("/v1/chat/completions", payload, timeout=300) elapsed = time.perf_counter() - started timings = response.get("timings", {}) usage = response.get("usage", {}) completion_tokens = int(timings.get("predicted_n", usage.get("completion_tokens", 0))) generation_seconds = float(timings.get("predicted_ms", 0)) / 1000.0 answer = demo.message_text(response) record = { "repeat": repeat, "prompt": prompt, "response": answer, "elapsed_seconds": round(elapsed, 3), "completion_tokens": completion_tokens, "prompt_tokens": int(usage.get("prompt_tokens", timings.get("prompt_n", 0))), "generation_seconds": round(generation_seconds, 3), "generation_tokens_per_second": round( completion_tokens / generation_seconds if generation_seconds > 0 else 0, 2 ), "prompt_tokens_per_second": round(float(timings.get("prompt_per_second", 0)), 2), "finish_reason": response.get("choices", [{}])[0].get("finish_reason"), } records.append(record) print( f"[{len(records):02d}/{len(PROMPTS) * REPEATS}] {prompt}\n" f" 回應：{answer}\n" f" 耗時：{elapsed:.3f}s | 生成：{record['generation_tokens_per_second']:.2f} tokens/s " f"({completion_tokens} tokens)\n", flush=True, ) speeds = [float(row["generation_tokens_per_second"]) for row in records] durations = [float(row["elapsed_seconds"]) for row in records] server_log = BENCH_LOG.read_text(encoding="utf-8", errors="replace") kv_mapping_seen = "backed by SSD file" in server_log and str(kv_path) in server_log weight_mmap_enabled = "loading model" in server_log and str(demo.MODEL_PATH) in server_log active_mappings = demo.mapped_project_files(process.pid) model_stats = demo.backing_file_stats(demo.MODEL_PATH) kv_stats = demo.backing_file_stats(kv_path) kv_mapping_present = any(path.endswith("kv-ssd.cache") for path in active_mappings) model_mapping_present = any(path.endswith(demo.MODEL_FILE) for path in active_mappings) summary = { "generated_at": datetime.now(timezone.utc).isoformat(), "model": f"{demo.MODEL_REPO}/{demo.MODEL_FILE}", "engine": "llama.cpp CPU", "llama_cpp_commit": demo.LLAMA_COMMIT, "hardware": { "platform": sys.platform, "machine": os.uname().machine, "cpu": cpu_name(), "logical_cpu_count": os.cpu_count(), "llama_threads": demo.THREADS, "context": demo.CONTEXT, }, "settings": { "weight_mapping": "GGUF file-backed mmap on NVMe; MAP_POPULATE disabled; madvise(MADV_DONTNEED) every 1 ms", "kv_cache": "K and V q8_0, forced SSD-file-backed mmap; cache mapped to ./models/kv-ssd.cache and madvise-swept", "system_pagecache_reclaim": "POSIX_FADV_DONTNEED on model and KV backing files every 0.1 s", "model_file": str(demo.MODEL_PATH.relative_to(demo.ROOT)), "kv_file": str(kv_path.relative_to(demo.ROOT)), "file_backing_stats": {"model": model_stats, "kv": kv_stats}, "kv_ssd_min_mib": 0, "mmap_release_interval_ms": demo.MMAP_RELEASE_INTERVAL_MS, "temperature": 0, "warmup_requests_excluded_from_records": 1, "repeats_per_prompt": REPEATS, "max_tokens": MAX_TOKENS, }, "system_swap_before": system_swap_before, "system_swap_after": demo.system_swap(), "memory": monitor.summary(), "pagecache_reclaimer": pagecache.summary(), "active_file_mappings": active_mappings, "ssd_activation_checks": { "weights_loaded_from_gguf_mmap": weight_mmap_enabled, "kv_ssd_backed_file_log_message": kv_mapping_seen, "kv_ssd_file_mapped_in_llama_server": kv_mapping_present, "model_gguf_mapped_in_llama_server": model_mapping_present, "model_and_kv_same_filesystem": model_stats["device_id"] == kv_stats["device_id"], "both_model_and_kv_present_in_process_maps": model_mapping_present and kv_mapping_present, }, "summary": { "measurement_count": len(records), "mean_generation_tokens_per_second": round(sum(speeds) / len(speeds), 2), "mean_end_to_end_seconds": round(sum(durations) / len(durations), 3), "min_generation_tokens_per_second": min(speeds), "max_generation_tokens_per_second": max(speeds), }, "results": records, "all_passed": bool( monitor.summary()["within_memory_limit"] and monitor.summary()["all_process_swap_samples_zero"] and kv_path.is_file() and model_mapping_present and kv_mapping_present and system_swap_before.get("used_bytes", 0) == 0 and demo.system_swap().get("used_bytes", 0) == 0 ), } demo.write_json(RESULT_PATH, summary) print(f"Benchmark JSON：{RESULT_PATH}") print(f"平均生成速度：{summary['summary']['mean_generation_tokens_per_second']} tokens/s") print(f"峰值 RSS：{summary['memory']['peak_rss_mib']} MiB") print(f"整機 swap 使用：{summary['system_swap_after'].get('used_bytes', 0)} bytes") return 0 finally: pagecache.stop() monitor.stop() if process.poll() is None: process.send_signal(signal.SIGTERM) try: process.wait(timeout=15) except subprocess.TimeoutExpired: process.kill() process.wait() if RESULT_PATH.exists(): summary = json.loads(RESULT_PATH.read_text(encoding="utf-8")) summary["memory"] = monitor.summary() summary["pagecache_reclaimer"] = pagecache.summary() summary["system_swap_after"] = demo.system_swap() summary["all_passed"] = bool( summary["memory"]["within_memory_limit"] and summary["memory"]["all_process_swap_samples_zero"] and summary["system_swap_before"].get("used_bytes", 0) == 0 and summary["system_swap_after"].get("used_bytes", 0) == 0 ) demo.write_json(RESULT_PATH, summary) if __name__ == "__main__": raise SystemExit(main())