diff --git a/models/experimental/MODEL_TIMING.ja.md b/models/experimental/MODEL_TIMING.ja.md new file mode 100644 index 0000000..c823e3d --- /dev/null +++ b/models/experimental/MODEL_TIMING.ja.md @@ -0,0 +1,56 @@ + +# 単一モデルの native 処理時間 + +`model_timing.py` はlegacy/identity/filteredの3モデルを同じharness・library・corpusで測定する。 +Pythonはbuild・入力選択・実行順制御だけを担当し、計時はC++の`steady_clock`で行う。 +全detector、Python API、複数候補競合、マルチスレッドのbenchmarkではない。 + +```sh +taskset -c 2 uv run --no-project python models/experimental/model_timing.py \ + /disk/identity-training.json /disk/filtered-training.json /disk/validation/manifest.json \ + build/release/src/libuchardet.a /disk/timing.json --iterations 20000 --repeats 7 +``` + +CPU番号は実行環境のallowed affinityから選ぶ。Linuxの1 CPU固定を要求し、未固定なら拒否する。 +CPU名・governorの前後値、OS/kernel、compiler/flags、library/model/input/binaryのhashを保存する。 +governorやboostを変更しない。affinityはCPUの占有、SMT相方の停止、周波数一定を保証しない。 +測定中は別のbuild/test等の重い作業を実行しない。 + +## 計測区間 + +入力を一度読み、scratch bufferとproberを一度確保した後、次をC++内で繰り返す。 + +1. prober reset +2. 同じ全文入力を既存filterへ渡す +3. 非空のfilter結果をproberへ一回feed +4. 内部confidence取得・bit変換・volatile checksum更新 + +入力I/O、process起動、Python、コンパイル、入力/scratch/proberの確保、JSON出力は区間外。 +128回のwarm-upも区間外。iteration数は1〜1,000,000、試行数は3〜31。 +各processは10秒timeout。時間不足なら勝手にiterationを減らしたreportを出さず失敗する。 + +全モデルを先にbuildし、文書・試行ごとに3モデルの実行順を循環させる。 +計測前の非計測観測と、各試行の最終counter/state/confidence/reset結果が一致することを検証。 +checksumはconfidence bits × iteration数とも照合する。 +既存filter/prober処理・table・公開APIは変更しない。 + +## 集計の意味 + +各試行の`elapsed_ns / iterations`から文書別median/min/maxを計算する。 +`p95_trial_mean_ns_per_iteration`は試行平均のnearest-rank p95であり、 +**リクエスト単位のp95 latencyではない**。7試行なら最大の試行平均になる。 + +`summed_document_trial_summary`は同じ試行番号の独立した文書別計時を合計したもの。 +各文書を同じ回数評価する比較用の値で、corpusを交互に巡回する処理を実測した値ではない。 +同じ文書を繰り返すためcacheが温まる。異なる文書が連続するingestion workloadへの +一般化には別benchmarkが必要である。 + +計時値は再実行で変動する。JSON全byteの一致を性能の再現性と扱わず、独立runの分布を比較する。 +保存先はrunごとに変える。異なる既存reportを上書きしない。 + +## 採用判断とは別 + +このtoolは単一モデルのreuse処理コストだけを測る。allocation数、ピークmemory、 +モデル生成時間、全detectorのthroughput/候補順位は未測定。 +新規モデルの品質・権利・confidence校正は別gateであり、速さだけで採用しない。 +CIの短い反復は計測機構と出力一致のtestで、性能数値を主張するものではない。 diff --git a/models/experimental/model_timing.py b/models/experimental/model_timing.py new file mode 100644 index 0000000..caed9f4 --- /dev/null +++ b/models/experimental/model_timing.py @@ -0,0 +1,182 @@ +# SPDX-License-Identifier: MIT +"""Pinned-CPU warmed native single-model timing; no Python or file I/O in measured interval.""" + +from __future__ import annotations + +import argparse +import json +import os +import platform +import statistics +import tempfile +from pathlib import Path + +import native_comparison +import sequence_probe +from model import canonical, digest, write_idempotent +from sequence_contract import content_hash + +NAMES = ("legacy", "identity", "filtered") + + +def summary(trials, iterations): + if not trials or any(type(value) is not int or value <= 0 for value in trials): + raise ValueError("positive integer trial times required") + if type(iterations) is not int or not 1 <= iterations <= 1000000: + raise ValueError("invalid iteration count") + values = sorted(value / iterations for value in trials) + return dict( + median_ns_per_iteration=statistics.median(values), + minimum_ns_per_iteration=values[0], + maximum_ns_per_iteration=values[-1], + # Nearest-rank p95 of trial averages, NOT per-request tail latency. + p95_trial_mean_ns_per_iteration=values[(95 * len(values) + 99) // 100 - 1], + ) + + +def affinity(): + if not hasattr(os, "sched_getaffinity"): + raise ValueError("this benchmark runner requires Linux CPU affinity") + cpus = sorted(os.sched_getaffinity(0)) + if len(cpus) != 1: + raise ValueError("pin the process to one allowed CPU with taskset") + return cpus + + +def cpu_environment(cpu): + info = {} + for block in Path("/proc/cpuinfo").read_text(encoding="utf-8").split("\n\n"): + fields = dict(line.split(":", 1) for line in block.splitlines() if ":" in line) + fields = {key.strip(): value.strip() for key, value in fields.items()} + if fields.get("processor") == str(cpu): + info["model_name"] = fields.get("model name") + break + governor = Path(f"/sys/devices/system/cpu/cpu{cpu}/cpufreq/scaling_governor") + info["scaling_governor"] = governor.read_text().strip() if governor.is_file() else None + return info + + +def run( + identity, filtered, manifest, root, split, library, iterations=20000, repeats=7, compiler="c++" +): + if type(iterations) is not int or not 1 <= iterations <= 1000000: + raise ValueError("iterations must be in [1, 1000000]") + if type(repeats) is not int or not 3 <= repeats <= 31: + raise ValueError("repeats must be in [3, 31]") + cpus = affinity() + cpu_info = cpu_environment(cpus[0]) + records = native_comparison.select_records(identity, filtered, manifest, root, split) + library = Path(library).resolve(strict=True) + library_hash = digest(library.read_bytes()) + profiles, binaries, baselines = {}, {}, {} + with tempfile.TemporaryDirectory(prefix="uchardet-model-timing-") as temporary: + for name, training in (("legacy", None), ("identity", identity), ("filtered", filtered)): + directory = Path(temporary) / name + directory.mkdir() + binary, provenance = ( + sequence_probe.build_reference(library, directory, compiler) + if training is None + else sequence_probe.build(training["contract"], library, directory, compiler) + ) + if provenance["static_library_sha256"] != library_hash: + raise ValueError("library differs between model builds") + binaries[name] = binary + baselines[name] = { + source["id"]: sequence_probe.observe(binary, data) for source, _, data in records + } + profiles[name] = dict( + provenance=provenance, + training_artifact_hash=training["content_hash"] if training else None, + sources={ + source["id"]: dict( + source=source, + sample_sha256=sample["sha256"], + byte_length=len(data), + elapsed_ns=[], + ) + for source, sample, data in records + }, + ) + # Build all models before timing. Rotate their order for every source/repetition. + for repeat in range(repeats): + for index, (source, _, data) in enumerate(records): + offset = (repeat + index) % len(NAMES) + for name in (*NAMES[offset:], *NAMES[:offset]): + observed = sequence_probe.observe(binaries[name], data, iterations) + timing = observed.pop("benchmark") + if observed != baselines[name][source["id"]]: + raise ValueError("timed run differs from untimed observation") + profiles[name]["sources"][source["id"]]["elapsed_ns"].append( + timing["elapsed_ns"] + ) + if affinity() != cpus or digest(library.read_bytes()) != library_hash: + raise ValueError("affinity/library changed during timing") + for profile in profiles.values(): + for source in profile["sources"].values(): + source["summary"] = summary(source["elapsed_ns"], iterations) + # Sum independent hot-document trials; do not present this as an interleaved corpus pass. + totals = [ + sum(source["elapsed_ns"][i] for source in profile["sources"].values()) + for i in range(repeats) + ] + profile["summed_document_trial_summary"] = summary(totals, iterations) + report = dict( + schema="native-model-timing-v1", + corpus_content_hash=manifest["content_hash"], + split=split, + scope="reused single prober: reset + filter + feed + confidence + checksum", + excluded="input I/O, allocation of input/scratch/prober, compilation, Python, output", + iterations=iterations, + repeats=repeats, + warmup_iterations=128, + clock="C++ steady_clock; nanoseconds", + cpu_affinity=cpus, + aggregate_policy=( + "sum of independent warmed same-document trial means; not interleaved corpus" + ), + environment=dict( + system=platform.system(), + release=platform.release(), + machine=platform.machine(), + cpu_before=cpu_info, + cpu_after=cpu_environment(cpus[0]), + ), + driver_dependencies={ + path.name: digest(path.read_bytes()) + for path in (Path(__file__), Path(native_comparison.__file__)) + }, + profiles=profiles, + ) + report["content_hash"] = content_hash(report) + return report + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + for name in ("identity", "filtered", "manifest", "library", "output"): + parser.add_argument(name, type=Path) + parser.add_argument("--split", choices=("tuning", "validation"), default="validation") + parser.add_argument("--iterations", type=int, default=20000) + parser.add_argument("--repeats", type=int, default=7) + parser.add_argument("--cxx", default="c++") + args = parser.parse_args() + + def load(path): + return json.loads(path.read_text(encoding="utf-8")) + + report = run( + load(args.identity), + load(args.filtered), + load(args.manifest), + args.manifest.parent, + args.split, + args.library, + args.iterations, + args.repeats, + args.cxx, + ) + write_idempotent(args.output, canonical(report)) + + +if __name__ == "__main__": + main() diff --git a/models/experimental/sequence-probe.cpp b/models/experimental/sequence-probe.cpp index 3036393..9d65866 100644 --- a/models/experimental/sequence-probe.cpp +++ b/models/experimental/sequence-probe.cpp @@ -2,12 +2,14 @@ // Independent observation harness. Detector/filter implementations stay in the library. #include "sequence-model.hpp" #include +#include #include #include #include #include #include #include +#include #include class SequenceProbe : public nsSingleByteCharSetProber { @@ -39,7 +41,17 @@ class SequenceProbe : public nsSingleByteCharSetProber { int main(int argc, char** argv) { try { - if (argc != 2) throw std::runtime_error("usage: sequence-probe FILE"); + if (argc != 2 && argc != 3) + throw std::runtime_error("usage: sequence-probe FILE [ITERATIONS]"); + std::uint64_t iterations = 0; + if (argc == 3) { + const std::string argument(argv[2]); + if (argument.empty() || argument.find_first_not_of("0123456789") != std::string::npos) + throw std::runtime_error("iterations must be unsigned decimal"); + iterations = std::stoull(argument); + if (iterations == 0 || iterations > 1000000) + throw std::runtime_error("iterations must be in [1, 1000000]"); + } std::ifstream input(argv[1], std::ios::binary); if (!input) throw std::runtime_error("cannot open input"); const std::size_t limit = 65536; @@ -50,14 +62,34 @@ int main(int argc, char** argv) { if (data.size() > limit) throw std::runtime_error("input exceeds 65536 byte diagnostic limit"); std::vector buffer(std::max(1, data.size())); PRUint32 retained = 0; - if (!data.empty()) { - nsCharSetProber::FilterWithoutEnglishLettersToBuffer( - data.data(), static_cast(data.size()), buffer.data(), retained); - } - if (retained > data.size()) throw std::runtime_error("unexpected filter length"); SequenceProbe probe(&uchardet_sequence_pilot::model); - // Match the SBCS group's empty-filter behavior. This is a single, non-reversed prober. - if (retained) probe.HandleData(buffer.data(), retained, nullptr, nullptr); + const auto run = [&]() { + probe.Reset(); + retained = 0; + if (!data.empty()) { + nsCharSetProber::FilterWithoutEnglishLettersToBuffer( + data.data(), static_cast(data.size()), buffer.data(), retained); + } + if (retained > data.size()) throw std::runtime_error("unexpected filter length"); + // Match the SBCS group's empty-filter behavior; a single non-reversed prober. + if (retained) probe.HandleData(buffer.data(), retained, nullptr, nullptr); + const float confidence = probe.GetConfidence(0); + std::uint32_t bits = 0; + std::memcpy(&bits, &confidence, sizeof(bits)); + return bits; + }; + run(); + std::int64_t elapsed_ns = 0; + volatile std::uint64_t checksum = 0; + const unsigned warmup = 128; + if (iterations) { + for (unsigned i = 0; i < warmup; ++i) checksum += run(); + checksum = 0; + const auto start = std::chrono::steady_clock::now(); + for (std::uint64_t i = 0; i < iterations; ++i) checksum += run(); + elapsed_ns = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start).count(); + } std::cout << "{\"schema\":\"sequence-native-probe-v1\",\"raw_bytes\":" << data.size() << ",\"filtered_bytes\":" << retained << ",\"model_encoding\":\"" << probe.GetCharSetName(0) @@ -68,6 +100,12 @@ int main(int argc, char** argv) { std::cout << ",\"after_reset\":"; probe.Reset(); probe.print(); + if (iterations) { + std::cout << ",\"benchmark\":{\"iterations\":" << iterations + << ",\"warmup_iterations\":" << warmup + << ",\"elapsed_ns\":" << elapsed_ns + << ",\"checksum\":" << checksum << '}'; + } std::cout << "}\n"; return std::cout ? 0 : 1; } catch (const std::exception& error) { diff --git a/models/experimental/sequence_probe.py b/models/experimental/sequence_probe.py index 5bef14f..d9db0f0 100644 --- a/models/experimental/sequence_probe.py +++ b/models/experimental/sequence_probe.py @@ -110,20 +110,30 @@ def _build(header, model_metadata, library, directory, compiler): return binary, provenance -def observe(binary, data): +def observe(binary, data, iterations=None): if len(data) > 65536: raise ValueError("probe input exceeds 65536 bytes") + if iterations is not None and (type(iterations) is not int or not 1 <= iterations <= 1000000): + raise ValueError("iterations must be an integer in [1, 1000000]") with tempfile.TemporaryDirectory(prefix="uchardet-sequence-input-") as temporary: path = Path(temporary) / "input.bin" path.write_bytes(data) - result = subprocess.run( - [str(binary), str(path)], check=True, capture_output=True, timeout=10 - ) + command = [str(binary), str(path)] + if iterations is not None: + command.append(str(iterations)) + result = subprocess.run(command, check=True, capture_output=True, timeout=10) observation = json.loads(result.stdout) if observation.get("schema") != "sequence-native-probe-v1" or observation["raw_bytes"] != len( data ): raise ValueError("unexpected native observation") + if iterations is not None: + benchmark = observation["benchmark"] + if benchmark["iterations"] != iterations or benchmark["warmup_iterations"] != 128: + raise ValueError("unexpected native timing policy") + expected = int(observation["snapshot"]["confidence_bits"], 16) * iterations + if benchmark["elapsed_ns"] <= 0 or benchmark["checksum"] != expected: + raise ValueError("invalid native elapsed time/checksum") return observation diff --git a/models/experimental/test_model_timing.py b/models/experimental/test_model_timing.py new file mode 100644 index 0000000..511651d --- /dev/null +++ b/models/experimental/test_model_timing.py @@ -0,0 +1,46 @@ +# SPDX-License-Identifier: MIT +import unittest +from unittest.mock import patch + +import model_timing + + +class ModelTimingTests(unittest.TestCase): + def test_trial_summary_and_nearest_rank(self): + result = model_timing.summary([300, 100, 200], 10) + self.assertEqual( + result, + dict( + median_ns_per_iteration=20, + minimum_ns_per_iteration=10, + maximum_ns_per_iteration=30, + p95_trial_mean_ns_per_iteration=30, + ), + ) + self.assertEqual( + model_timing.summary(list(range(1, 21)), 1)["p95_trial_mean_ns_per_iteration"], 19 + ) + + def test_bad_times_and_iteration_counts(self): + for values in ([], [0], [-1], [True], [1.5]): + with self.assertRaises(ValueError): + model_timing.summary(values, 1) + for count in (0, -1, 1000001, True): + with self.assertRaises(ValueError): + model_timing.summary([1], count) + + def test_requires_single_cpu(self): + with patch.object(model_timing.os, "sched_getaffinity", return_value={2, 3}, create=True): + with self.assertRaisesRegex(ValueError, "taskset"): + model_timing.affinity() + with patch.object(model_timing.os, "sched_getaffinity", return_value={2}, create=True): + self.assertEqual(model_timing.affinity(), [2]) + + def test_runner_configuration_rejected_before_corpus_or_compile(self): + for iterations, repeats in ((0, 7), (1, 2), (1, 32), (True, 7), (1, True)): + with self.assertRaises(ValueError): + model_timing.run(None, None, None, None, None, None, iterations, repeats) + + +if __name__ == "__main__": + unittest.main() diff --git a/models/experimental/test_sequence_probe.py b/models/experimental/test_sequence_probe.py index 688d812..e930c73 100644 --- a/models/experimental/test_sequence_probe.py +++ b/models/experimental/test_sequence_probe.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: MIT import os import struct +import subprocess import tempfile import unittest from pathlib import Path @@ -12,6 +13,13 @@ class SequenceProbeGuards(unittest.TestCase): + def test_iteration_limits_before_execution(self): + for iterations in (0, -1, 1000001, True, 1.5): + with patch.object(sequence_probe.subprocess, "run") as run: + with self.assertRaisesRegex(ValueError, "iterations"): + sequence_probe.observe(Path("unused"), b"text", iterations) + run.assert_not_called() + def test_input_limit_before_native_execution(self): with patch.object(sequence_probe.subprocess, "run") as run: with self.assertRaisesRegex(ValueError, "65536"): @@ -108,6 +116,27 @@ def test_repeat_and_build_provenance(self): self.contract, Path(os.environ["UCHARDET_STATIC_LIBRARY"]), self.temporary.name ) + def test_timing_preserves_observation_and_checksum(self): + for data in (b"", b"ASCII", b"caf\xe9", b"\xe9\xe8\xe0"): + baseline = self.observed(data) + timed = sequence_probe.observe(self.binary, data, iterations=3) + benchmark = timed.pop("benchmark") + self.assertEqual(timed, baseline) + self.assertEqual(benchmark["iterations"], 3) + self.assertEqual(benchmark["warmup_iterations"], 128) + self.assertGreater(benchmark["elapsed_ns"], 0) + self.assertEqual( + benchmark["checksum"], 3 * int(baseline["snapshot"]["confidence_bits"], 16) + ) + + def test_native_iteration_argument_validation(self): + for argument in ("0", "-1", "1x", "1000001"): + result = subprocess.run( + [str(self.binary), "unused", argument], capture_output=True, text=True, timeout=10 + ) + self.assertNotEqual(result.returncode, 0) + self.assertIn("iterations", result.stderr) + if __name__ == "__main__": unittest.main()