検索インデックスのhash table化

1 files changed, 130 insertions, 1 deletions

diff --git a/scripts/python/benchmark.py b/scripts/python/benchmark.py
index 69219c9..5a146ce 100644
--- a/scripts/python/benchmark.py
+++ b/scripts/python/benchmark.py
@@ -39,9 +39,12 @@ from mcp.types import CallToolResult
 # ---------------------------------------------------------------------------
 DEFAULT_ITERATIONS = 100
 DEFAULT_CONTENT_MEDIUM = 1024  # bytes
+DEFAULT_CONTENT_8K = 8 * 1024  # bytes
 DEFAULT_CONTENT_LARGE = 32 * 1024  # bytes
+DEFAULT_CONTENT_MAX = 1024 * 1024  # bytes
 WARMUP_COUNT = 5
 REBUILD_ITERATIONS = 5
+P1_CORPUS_MESSAGES = 20
 
 ROOT = Path(__file__).resolve().parent.parent.parent
 DEFAULT_RELEASE_EXE = (
@@ -90,6 +93,7 @@ class BenchmarkReport:
     iterations: int
     warmup: int
     scenarios: list[ScenarioResult] = field(default_factory=list)
+    domain_index: list[dict[str, Any]] = field(default_factory=list)
     health_after: dict[str, Any] | None = None
     schema_check: dict[str, Any] | None = None
     errors: list[str] = field(default_factory=list)
@@ -142,6 +146,25 @@ def summarize(name: str, samples: list[LatencySample]) -> ScenarioResult:
     return result
 
 
+def single_sample(name: str, duration_ms: float, notes: str) -> ScenarioResult:
+    return ScenarioResult(
+        name=name,
+        iterations=1,
+        successes=1,
+        failures=0,
+        samples=[duration_ms],
+        throughput_ops_per_sec=1000.0 / duration_ms if duration_ms > 0 else 0.0,
+        latency_p50_ms=duration_ms,
+        latency_p95_ms=duration_ms,
+        latency_p99_ms=duration_ms,
+        latency_min_ms=duration_ms,
+        latency_max_ms=duration_ms,
+        latency_mean_ms=duration_ms,
+        total_duration_ms=duration_ms,
+        notes=notes,
+    )
+
+
 def text_content(result: CallToolResult) -> str:
     if result.isError:
         raise RuntimeError(f"tool returned isError: {result.content!r}")
@@ -168,6 +191,24 @@ def make_metadata_json(tags: list[str] | None = None, tier: str = "short") -> st
     return json.dumps(obj, separators=(",", ":"))
 
 
+def run_domain_index_benchmarks(server: Path) -> list[dict[str, Any]]:
+    if len(server.parents) < 3:
+        return []
+    benchmark = server.parents[2] / "tests" / server.parent.name / "tiny-convmem-search-benchmark.exe"
+    if not benchmark.exists():
+        return []
+    results: list[dict[str, Any]] = []
+    for size, iterations in ((1024, 100), (8192, 100), (32768, 50), (1048576, 5)):
+        completed = subprocess.run(
+            [str(benchmark), str(size), str(iterations)],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        results.append(json.loads(completed.stdout))
+    return results
+
+
 # ---------------------------------------------------------------------------
 # クライアント・サーバランチャー
 # ---------------------------------------------------------------------------
@@ -377,6 +418,64 @@ class ScenarioRunner:
             "write_message (large 32KiB)",
         )
 
+    async def run_write_8k(self, iterations: int) -> ScenarioResult:
+        content = random_content(DEFAULT_CONTENT_8K)
+        return await self.run_write(
+            content,
+            make_metadata_json(),
+            iterations,
+            "write_message (P1 8KiB)",
+        )
+
+    async def run_write_max(self, iterations: int) -> ScenarioResult:
+        content = random_content(DEFAULT_CONTENT_MAX)
+        return await self.run_write(
+            content,
+            make_metadata_json(),
+            iterations,
+            "write_message (P1 1MiB)",
+        )
+
+    async def run_p1_startup_rebuild(
+        self, size: int
+    ) -> tuple[ScenarioResult, ScenarioResult]:
+        data_dir = self.new_data_dir()
+        session_id = f"p1-{size}"
+        content = random_content(size)
+        try:
+            async with ServerContext(self.exe, data_dir) as session:
+                for _ in range(P1_CORPUS_MESSAGES):
+                    await write_message(session, session_id, content)
+
+            started = time.perf_counter()
+            async with ServerContext(self.exe, data_dir) as session:
+                startup_ms = (time.perf_counter() - started) * 1000.0
+                health = await get_system_health(session)
+                if health.get("active_messages") != P1_CORPUS_MESSAGES:
+                    raise RuntimeError(f"P1 corpus count mismatch: {health}")
+
+                rebuild_started = time.perf_counter()
+                payload = await rebuild_index(session)
+                rebuild_ms = (time.perf_counter() - rebuild_started) * 1000.0
+                if payload.get("status") != "OK":
+                    raise RuntimeError(f"P1 rebuild failed: {payload}")
+
+            label = f"{size // 1024}KiB" if size < 1024 * 1024 else "1MiB"
+            return (
+                single_sample(
+                    f"startup index build (P1 {label} x{P1_CORPUS_MESSAGES})",
+                    startup_ms,
+                    "process start + initialize; corpus is loaded from the persisted log",
+                ),
+                single_sample(
+                    f"rebuild_index (P1 {label} x{P1_CORPUS_MESSAGES})",
+                    rebuild_ms,
+                    "compaction and complete in-memory search index rebuild",
+                ),
+            )
+        finally:
+            self.cleanup(data_dir)
+
     async def run_write_with_tags(self, iterations: int) -> ScenarioResult:
         content = "tagged message"
         metadata_json = make_metadata_json(tags=["benchmark", "tagged"], tier="short")
@@ -686,6 +785,13 @@ def print_report(report: BenchmarkReport) -> None:
         print("\nfinal health:")
         for k, v in report.health_after.items():
             print(f"  {k}: {v}")
+    if report.domain_index:
+        print("\ndomain search index:")
+        for item in report.domain_index:
+            print(
+                f"  {item['size_bytes']:>7} bytes: {item['mean_ms']:.3f} ms, "
+                f"peak working set {item['peak_working_set_bytes']} bytes"
+            )
 
 
 def write_json(report: BenchmarkReport, path: Path) -> None:
@@ -722,6 +828,17 @@ def write_markdown(report: BenchmarkReport, json_path: Path, md_path: Path) -> N
         lines.append(json.dumps(report.schema_check, indent=2, ensure_ascii=False))
         lines.append("\n```\n\n")
 
+    if report.domain_index:
+        lines.append("## Domain検索index構築\n\n")
+        lines.append("| payload | iterations | mean[ms] | peak working set[bytes] |\n")
+        lines.append("|---:|---:|---:|---:|\n")
+        for item in report.domain_index:
+            lines.append(
+                f"| {item['size_bytes']} | {item['iterations']} | {item['mean_ms']:.3f} | "
+                f"{item['peak_working_set_bytes']} |\n"
+            )
+        lines.append("\n")
+
     if report.health_after:
         lines.append("## 最終 health\n\n")
         lines.append("```json\n")
@@ -794,13 +911,25 @@ async def amain(argv: list[str]) -> int:
         iterations=args.iterations,
         warmup=args.warmup,
     )
+    report.domain_index = run_domain_index_benchmarks(exe.resolve())
 
     # シナリオ実行
     report.scenarios.append(await runner.run_write_small(args.iterations))
     report.scenarios.append(await runner.run_write_medium(args.iterations))
+    report.scenarios.append(await runner.run_write_8k(min(args.iterations, 100)))
     report.scenarios.append(await runner.run_write_large(args.iterations))
+    report.scenarios.append(await runner.run_write_max(min(args.iterations, 5)))
     report.scenarios.append(await runner.run_write_with_tags(args.iterations))
 
+    for payload_size in (
+        DEFAULT_CONTENT_MEDIUM,
+        DEFAULT_CONTENT_8K,
+        DEFAULT_CONTENT_LARGE,
+        DEFAULT_CONTENT_MAX,
+    ):
+        startup_result, p1_rebuild_result = await runner.run_p1_startup_rebuild(payload_size)
+        report.scenarios.extend((startup_result, p1_rebuild_result))
+
     report.scenarios.append(await runner.run_read(args.iterations))
 
     search_global, search_scoped = await runner.run_search(args.iterations)
@@ -859,4 +988,4 @@ def main() -> int:
 
 
 if __name__ == "__main__":
-    sys.exit(main())
\ No newline at end of file
+    sys.exit(main())