File size: 7,618 Bytes
6ab8274 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | """정정된 Math Ink 0.6 composite online/raster CPU 지연을 대표 입력으로 측정한다."""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
import json
import math
from pathlib import Path
import statistics
import sys
import time
import torch
PROJECT_ROOT = Path(__file__).parents[1]
SOURCE_ROOT = PROJECT_ROOT / "src"
for path in (PROJECT_ROOT, SOURCE_ROOT):
if str(path) not in sys.path:
sys.path.insert(0, str(path))
from math_grid_drawer.research.math_ink_06 import MathInk06Engine
from scripts.export_math_ink_06_litert import _load_representative_inputs06
def percentile_nearest_rank06(values: list[float], percentile: float) -> float:
"""필요 변수: 측정값·0~1 percentile. 작동 원리: 모바일 p95와 동일한 nearest-rank 값을 반환한다."""
if not values or not 0.0 <= percentile <= 1.0:
raise ValueError("percentile 입력이 유효하지 않습니다.")
ordered = sorted(values)
index = max(0, min(len(ordered) - 1, math.ceil(percentile * len(ordered)) - 1))
return float(ordered[index])
def module_state_bytes06(*modules: torch.nn.Module) -> int:
"""필요 변수: model·adapter module. 작동 원리: 중복 storage를 한 번만 세어 실제 tensor state bytes를 계산한다."""
seen: set[tuple[int, int]] = set()
total = 0
for module in modules:
for tensor in [*module.parameters(), *module.buffers()]:
storage = tensor.untyped_storage()
key = (storage.data_ptr(), storage.nbytes())
if key in seen:
continue
seen.add(key)
total += storage.nbytes()
return total
def _rss_bytes06() -> int | None:
"""필요 변수: 없음. 작동 원리: psutil이 있으면 현재 process RSS를 반환하고 없으면 명시적으로 결측 처리한다."""
try:
import psutil
except ImportError:
return None
return int(psutil.Process().memory_info().rss)
def _measure06(callable_, inputs: list[tuple[torch.Tensor, ...]], warmup: int) -> dict:
"""필요 변수: 고정 inference callable·대표 입력·warmup. 작동 원리: 표본별 wall latency와 output checksum을 측정한다."""
with torch.inference_mode():
for arguments in inputs[:max(1, min(warmup, len(inputs)))]:
callable_(*arguments)
latencies, checksum = [], 0
observed_rss = _rss_bytes06()
for arguments in inputs:
started = time.perf_counter()
output = callable_(*arguments)
latencies.append((time.perf_counter() - started) * 1000.0)
primary = output[0] if isinstance(output, tuple) else output
checksum = (checksum * 131 + int(primary.argmax(dim=-1)[0])) % 2_147_483_647
current_rss = _rss_bytes06()
if current_rss is not None:
observed_rss = max(observed_rss or 0, current_rss)
return {
"samples": len(latencies),
"mean_ms": statistics.fmean(latencies),
"p50_ms": percentile_nearest_rank06(latencies, 0.50),
"p95_ms": percentile_nearest_rank06(latencies, 0.95),
"maximum_ms": max(latencies),
"output_checksum": checksum,
"observed_process_rss_bytes": observed_rss,
}
def main() -> None:
"""필요 변수: composite artifact·대표 cache. 작동 원리: thread별 두 inference 경로를 독립 측정해 JSON으로 남긴다."""
parser = argparse.ArgumentParser(description="Benchmark Math Ink 0.6 composite CPU")
parser.add_argument("--checkpoint", type=Path, required=True)
parser.add_argument("--adapter-checkpoint", type=Path, required=True)
parser.add_argument("--representative-inputs", type=Path, required=True)
parser.add_argument("--threads", type=int, action="append", default=None)
parser.add_argument("--samples", type=int, default=76)
parser.add_argument("--warmup", type=int, default=5)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
requested_threads = args.threads or [1, 2, 4]
if any(value <= 0 for value in requested_threads):
raise ValueError("CPU thread는 양수여야 합니다.")
torch.set_num_interop_threads(1)
engine = MathInk06Engine(
args.checkpoint, adapter_checkpoint=args.adapter_checkpoint, device="cpu",
)
online_inputs, raster_inputs = _load_representative_inputs06(args.representative_inputs)
online_inputs = online_inputs[:args.samples]
raster_inputs = raster_inputs[:args.samples]
def online_forward(sequence: torch.Tensor):
"""필요 변수: canonical sequence. 작동 원리: 실제 runtime online composite branch를 호출한다."""
return engine.model.forward_online(engine.online_adapter(sequence))
def raster_forward(raster: torch.Tensor):
"""필요 변수: raster. 작동 원리: 실제 runtime virtual stroke·raster adapter·fusion을 호출한다."""
output = engine._forward_raster_composite06(raster)
return engine.fuse_raster_output(output)[0]
baseline_rss = _rss_bytes06()
rows = []
for thread_count in requested_threads:
torch.set_num_threads(thread_count)
online = _measure06(online_forward, online_inputs, args.warmup)
raster = _measure06(raster_forward, raster_inputs, args.warmup)
rows.append({
"threads": thread_count,
"online": online,
"raster": raster,
"proxy_gates": {
"online_p95_le_50ms": online["p95_ms"] <= 50.0,
"raster_p95_le_200ms": raster["p95_ms"] <= 200.0,
},
})
observed_rss_values = [
int(metrics["observed_process_rss_bytes"])
for row in rows for metrics in (row["online"], row["raster"])
if metrics["observed_process_rss_bytes"] is not None
]
maximum_observed_rss = max(observed_rss_values) if observed_rss_values else None
inference_rss_growth = (
max(0, maximum_observed_rss - baseline_rss)
if maximum_observed_rss is not None and baseline_rss is not None else None
)
state_bytes = module_state_bytes06(engine.model, engine.composite_adapter)
model_plus_inference = (
state_bytes + inference_rss_growth
if inference_rss_growth is not None else None
)
report = {
"schema": "aiflow-math-ink-06-composite-cpu-benchmark-v1",
"generated_at": datetime.now(timezone.utc).isoformat(),
"torch_version": torch.__version__,
"platform": sys.platform,
"model_version": engine.model_version,
"model_state_bytes": state_bytes,
"baseline_process_rss_bytes": baseline_rss,
"maximum_observed_process_rss_bytes": maximum_observed_rss,
"inference_rss_growth_bytes": inference_rss_growth,
"model_state_plus_inference_growth_bytes": model_plus_inference,
"memory_proxy_gate_le_100mib": (
model_plus_inference <= 100 * 1024 * 1024
if model_plus_inference is not None else None
),
"rows": rows,
"interpretation_limit": (
"Windows PyTorch CPU proxy이며 Android LiteRT·배터리·delegate 성능 판정이 아니다."
),
"product_validation": False,
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8",
)
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
|