diff options
| -rw-r--r-- | docs/overnight-status.md | 7 | ||||
| -rw-r--r-- | docs/维护说明.md | 41 | ||||
| -rwxr-xr-x | scripts/m3_supervise.py | 39 | ||||
| -rw-r--r-- | src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj | 1 | ||||
| -rw-r--r-- | src/LivingVillage.Desktop.Tests/M3BatchOutputTests.fs | 84 | ||||
| -rw-r--r-- | src/LivingVillage.Headless/BatchOutput.fs | 75 | ||||
| -rw-r--r-- | src/LivingVillage.Headless/LivingVillage.Headless.fsproj | 1 | ||||
| -rw-r--r-- | src/LivingVillage.Headless/Program.fs | 93 |
8 files changed, 313 insertions, 28 deletions
diff --git a/docs/overnight-status.md b/docs/overnight-status.md index e8c2885..2896986 100644 --- a/docs/overnight-status.md +++ b/docs/overnight-status.md @@ -1191,3 +1191,10 @@ Updated: 2026-09-20 02:54:20 CST - **P14 派发**：prompt_async 204（13:03 CST 起 loop step 推进中，providerID=ark-plan modelID=deepseek-v4.1-flash 日志证实）。范围=上述 4 项整改+evidence 截图更新+维护说明补充；P13 已过验收部分别动。分支 p14-art-fix，verdict 协议照旧。 - **M3 phase2 长测**：lv-m3-phase2.service MainPID=1694385 运行 ~10.5h，CPU 累计 1d13h43m 持续计算；日志仍 0 world 行无 exit 文件，verdict=null——运行中不计入验收。 - completion-notice.md 不存在。待收口：P14 复验合并、M3 phase2 终态、最终发布门。 + +## 2026-09-21 22:2x CST — M3-1 批验可观测性 + 门槛缩尺 + 长局成本量化（分支 m3-observability，未推送） +- **流式输出**：`--batch K D` 改为每个世界完成即打印 `world=... OK elapsed_s=<秒>` 与 `[done k/N] elapsed=…s world=…` 并 flush；`batch_summary` 旧字段保留、追加 `workers=`/`wall_s=`。实测 `LV_BATCH_WORKERS=4 --batch 4 10`：world 行分别在 368.94s / 431.60s / 460.82s / 545.22s 出现（完成序 3,0,2,1），summary 545.23s，`M3_ACCEPTANCE=PASS`，exit 0；证据 `evidence/stream/batch-4x10.log`（逐行时间戳）。 +- **门槛缩尺**：`scripts/m3_supervise.py` `PHASES.phase2` 由 `(50,100)` 改为 `(50,20)`；`--selftest` 仍 `SELFTEST=PASS`。理由与实测（phase2 11.4h 仅 6/50、100 天≈6.8 CPU-h、全量≈335 CPU-h）写入 `docs/维护说明.md`。判据语义 check a/b/c 未改。 +- **成本探针**：新增只读 `--cost-probe [D1,D2,...]`（默认 1,5,10,20,50,100）。本机单 worker、npcs=30、seed=42 实测：1d 21.615s/19.0GB/gen0=2272，5d 111.645s/98.1GB/gen0=11728，10d 226.226s/202.4GB/gen0=24203（约 20.2GB/天·世界）。证据 `evidence/stream/cost-probe-1-5-10.log`。 +- **长局机理（只测不改）**：`World.Rumors` 永不裁剪，实测 count 1d=2025 / 5d=9821 / 10d=19493（≈1950/天），而 `latestRumorFor`/`rumorIsDuplicate`/`rumorPath` 每次全表扫描 → 单次交互成本随年龄线性、整轮超线性；`Mind.Memory` 有上限（恒 64）非主因。修复方向（裁剪工作集/Map 索引/限窗）留下一单，本单不实施。证据 `evidence/stream/growth-1-5-10.log`。 +- **待收口**：由 Hermes 独立复验本分支；正式 M3 重跑走冻结流程（50×20）。macOS 打包脚本事后单已排队（等本单交付）。 diff --git a/docs/维护说明.md b/docs/维护说明.md index 4635b6d..19ca201 100644 --- a/docs/维护说明.md +++ b/docs/维护说明.md @@ -1,6 +1,6 @@ # 维护说明 -结论先行：主分支保持 0 警告 0 错误 + Desktop 79/79、Kernel 79/79 全绿；性能基线 +结论先行：主分支保持 0 警告 0 错误 + Desktop 86/86、Kernel 79/79 全绿；性能基线 final_digest 固定 953775FAEB2FDDE97289491AA260BD8D390C571E48A7A13AD2CB6FB7124F7F6C。 本文记录模块职责、不变量与逐字可复制的验收命令。 @@ -83,6 +83,40 @@ dotnet src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll - digest 变化处理：先比对 `WorldSave.save world` 全文文本（版本头变化不构成失败即 digest 规范化后等价）； 再查 Sim.step 数值路径是否被改动（不允许）。历史归属见 `artifacts/perf-independent/*/README.md`。 +## M3 批验：流式可观测性、门槛与长局成本（M3-1） + +- **流式输出**：`--batch K D` 每个世界一完成就立即打印 `world=... OK elapsed_s=<秒>` 与 + `[done k/N] elapsed=…s world=…` 并 flush，结束时 `batch_summary` 旧字段原样保留并追加 + `workers=` 与 `wall_s=`（总墙钟）。完成序可能非升序，监督器按索引集合恰为 0..K-1 校验。 +- **正式门槛缩尺 50×100 → 50×20**（`scripts/m3_supervise.py` 的 `PHASES.phase2`）： + phase2（50×100）实测 11.4 小时仅完成 6/50 世界，单世界 100 天约 6.8 CPU 小时， + 全量约 335 CPU 小时，不可行（诊断见 + `artifacts/m3-acceptance/final-candidate-20260921T095500Z/logs/phase2_20260921T101500.log.diagnosis.txt`）。 + Hermes 实测单 worker 成本：1 天 25.6s / 5 天 123s / 10 天 254s / 20 天 ≈900s / + 100 天 ≈6.8 CPU-h。缩尺只改规模，判据语义 check a/b/c 未动。 +- **成本探针**：`--cost-probe [D1,D2,...]`（默认 `1,5,10,20,50,100`）逐段真实测量单世界 + `ticks_per_s`、墙钟、`allocated_bytes` 与 gen0/1/2，不估算、不写死。 + 本机（单 worker，npcs=30，seed=42）实测： + + | days | elapsed_s | ticks_per_s | allocated_bytes | gen0 | gen1 | gen2 | + |---|---|---|---|---|---|---| + | 1 | 21.615 | 239837.5 | 19,001,478,648 | 2272 | 286 | 11 | + | 5 | 111.645 | 232165.0 | 98,066,613,520 | 11728 | 11 | 2 | + | 10 | 226.226 | 229151.8 | 202,369,225,952 | 24203 | 22 | 2 | + + 即约 **20.2 GB / 天·世界** 的分配；10 天内吞吐近乎平稳，超线性主要体现在更老的世界上 + （Hermes 20 天 ≈45 s/天、100 天 ≈245 s/天）。 +- **成本随年龄上涨的机理（只测不改）**：`World.Rumors: RumorEvent list` 永不裁剪， + 每次谣言扩散 `rumor :: world.Rumors` 前插。实测 `rumor_trace count`：1 天 2025、 + 5 天 9821、10 天 19493（≈1950/天线性累积）。而 `latestRumorFor` / `rumorIsDuplicate` / + `rumorPath` 每次都 `world.Rumors |> List.filter/tryFind` 全表扫描，因此每次聊天/对话的 + 成本随世界年龄线性增长、整轮成本随时长超线性（≈平方）增长；同时每次扫描与前插都产生 + 列表分配，推高 GC。`Mind.Memory` 有 `memoryCapacity` 上限（实测恒 64），不是主因。 +- **修复方向候选（本单不实施，避免动确定性语义）**：①按 `rumorFreshnessTicks`/半衰期 + 确定性裁剪工作集（谱系/annal 另存）；②以 `Map<RumorId, RumorEvent>` 或逐接收者索引替代 + 全表扫描；③把 `latestRumorFor`/`rumorIsDuplicate` 限制到最近窗口。以上都需单开一单、 + 重跑 performance-baseline 与全部 digest 后由 Hermes 走冻结流程。 + ## 已验证命令 ```bash @@ -96,6 +130,11 @@ dotnet build LivingVillage.sln -c Release 2>&1 | tail -n3 dotnet test src/LivingVillage.Desktop.Tests -c Release --no-build dotnet test src/LivingVillage.Kernel.Tests -c Release --no-build +# M3 批次（流式；判据行与原格式逐字节兼容，仅行尾追加 elapsed_s） +LV_BATCH_WORKERS=4 dotnet src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll --batch 4 10 +# 长局成本阶梯（只读，真实测量） +dotnet src/LivingVillage.Headless/bin/Release/net8.0/LivingVillage.Headless.dll --cost-probe 1,5,10 + # 素材再生成（确定性，输出 byte-identical） python3 scripts/make-jiangnan-art.py && python3 scripts/make-cc0-art.py diff --git a/scripts/m3_supervise.py b/scripts/m3_supervise.py index e5a2315..26dfb4a 100755 --- a/scripts/m3_supervise.py +++ b/scripts/m3_supervise.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """M3 验收监督（加固版）：供 Hermes cron 每 5 分钟调用（no_agent）。 -状态机: idle -> phase1(--batch 30 20) -> phase2(--batch 50 100) -> done +状态机: idle -> phase1(--batch 30 20) -> phase2(--batch 50 20) -> done - flock 单实例防重入；抢不到锁静默退出 - 冻结运行: 首次启动把 Release 输出目录递归复制到 scripts/m3_frozen/<stamp>_<commit>/， v2 清单覆盖全部子目录文件 sha256；两阶段都从冻结副本运行，后续构建不会覆盖 @@ -11,7 +11,8 @@ - PASS 判定收紧（全部满足才 PASS，任一不满足即 FAIL/停机）: 1) exitfile 退出码 == 0 2) `batch worlds=K days=D` 与固定 PHASES[phase] 完全一致（不用 state 里可篡改值） - 3) `world=` 行: 行数==K、索引恰为 0..K-1 无重复无缺失、seed==42+i + 3) `world=` 行: 行数==K、索引恰为 0..K-1 无重复无缺失（允许流式完成序）、seed==42+i； + 行尾 `elapsed_s=<秒>` 为追加观测字段，不参与判据 4) `batch_summary worlds=K passed=K failed=0` 5) 行 `M3_ACCEPTANCE=PASS` 6) 冻结清单 sha256 全部一致 + commit 绑定；v2 清单另查额外/缺失文件 @@ -45,7 +46,7 @@ STATE_PATH = ROOT / "scripts/m3_state.json" LOCK_PATH = ROOT / "scripts/m3_supervise.lock" LOG_DIR = ROOT / "scripts/m3_logs" FROZEN_ROOT = ROOT / "scripts/m3_frozen" -PHASES = {"phase1": (30, 20), "phase2": (50, 100)} +PHASES = {"phase1": (30, 20), "phase2": (50, 20)} NEXT = {"phase1": "phase2", "phase2": None} FRAG = "LivingVillage.Headless.dll" @@ -239,6 +240,14 @@ def read_exitcode(exitfile): return None +def world_line_verdict(line): + """world 行的判据 token；忽略行尾追加的 `elapsed_s=<秒>` 观测字段。""" + tokens = line.split() + while tokens and re.fullmatch(r"elapsed_s=\d+(\.\d+)?", tokens[-1]): + tokens.pop() + return tokens[-1] if tokens else "" + + def verify_batch(text, worlds, days): """完整批次判据校验（固定 worlds/days 由调用方从 PHASES 取）。 返回问题列表（空=满足全部收紧条件）。""" @@ -263,7 +272,8 @@ def verify_batch(text, worlds, days): if malformed: problems.append(f"malformed world lines: {malformed[:5]}") idxs = [i for i, _ in parsed] - if idxs != list(range(worlds)): + # 流式输出允许完成序非升序，判据是索引集合恰为 0..K-1（无重无缺）。 + if sorted(idxs) != list(range(worlds)): dup = sorted({i for i in idxs if idxs.count(i) > 1}) missing = sorted(set(range(worlds)) - set(idxs)) extra = sorted(set(idxs) - set(range(worlds))) @@ -271,7 +281,7 @@ def verify_batch(text, worlds, days): badseed = [i for i, s in parsed if s != 42 + i] if badseed: problems.append(f"seed binding mismatch (expect seed=42+i): worlds={badseed[:8]}") - bad = [l.split()[0] for l in wlines if not l.split() or l.split()[-1] != "OK"] + bad = [l.split()[0] for l in wlines if world_line_verdict(l) != "OK"] if bad: problems.append(f"non-OK world lines: {bad[:8]}{'...' if len(bad) > 8 else ''}") s = re.search(r"^batch_summary worlds=(\d+) passed=(\d+) failed=(\d+) ", text, re.M) @@ -405,7 +415,7 @@ def decide(state, alive): state["verdict"] = "PASS" if nxt is None: state["phase"] = "done" - return "save", state, f"{phase} PASS (commit={state.get('commit')}, frozen={state.get('frozen')}) -> M3 acceptance complete (30x20 + 50x100)" + return "save", state, f"{phase} PASS (commit={state.get('commit')}, frozen={state.get('frozen')}) -> M3 acceptance complete (30x20 + 50x20)" return f"launch:{nxt}", state, None state["verdict"] = "FAIL" state["phase"] = "failed" @@ -641,6 +651,19 @@ def selftest(): "batch worlds=2 days=1 x\nworld=0 seed=42 a OK\nworld=1 seed=43 a OK\nbatch_summary worlds=2 passed=1 failed=1 x\nM3_ACCEPTANCE=PASS\n", 2, 1))) check("verify missing verdict token", any("M3_ACCEPTANCE=PASS" in p for p in verify_batch( "batch worlds=1 days=1 x\nworld=0 seed=42 a OK\nbatch_summary worlds=1 passed=1 failed=0 x\nM3_ACCEPTANCE=FAIL\n", 1, 1))) + # 流式输出：完成序非升序 + 行尾 elapsed_s 追加字段，均应通过判据。 + check("verify out-of-order streaming worlds", verify_batch( + "batch worlds=3 days=1 x\nworld=2 seed=44 a OK elapsed_s=3.1\nworld=0 seed=42 a OK elapsed_s=1.2\n" + "world=1 seed=43 a OK elapsed_s=2.0\nbatch_summary worlds=3 passed=3 failed=0 x\nM3_ACCEPTANCE=PASS\n", 3, 1) == []) + check("verify elapsed suffix tolerated with progress lines", verify_batch( + "batch worlds=2 days=1 x\n[done 1/2] elapsed=0.9s world=1\nworld=1 seed=43 a OK elapsed_s=0.9\n" + "[done 2/2] elapsed=1.2s world=0\nworld=0 seed=42 a OK elapsed_s=1.2\n" + "batch_summary worlds=2 passed=2 failed=0 x\nM3_ACCEPTANCE=PASS\n", 2, 1) == []) + check("verify elapsed suffix does not mask BAD", any("non-OK" in p for p in verify_batch( + "batch worlds=1 days=1 x\nworld=0 seed=42 a BAD elapsed_s=1.0\n" + "batch_summary worlds=1 passed=0 failed=1 x\nM3_ACCEPTANCE=FAIL\n", 1, 1))) + check("world_line_verdict ignores elapsed", world_line_verdict("world=0 seed=42 a OK elapsed_s=1.0") == "OK" + and world_line_verdict("world=0 seed=42 a BAD elapsed_s=1.0") == "BAD") # ---- 冻结清单（合成目录, v2 递归 + 额外/缺失检测） ---- with tempfile.TemporaryDirectory() as td: @@ -741,8 +764,8 @@ def selftest(): a, s, m = decide(dict(base, phase="phase2", log=str(plog)), lambda pid, st, fr: False) check("phase2 with phase1 log -> FAIL (fixed PHASES)", s["phase"] == "failed" and s["verdict"] == "FAIL") plog2 = td / "p2.log" - plog2.write_text(make_batch_log(50, 100)) - a, s, m = decide(dict(base, phase="phase2", worlds=50, days=100, log=str(plog2)), lambda pid, st, fr: False) + plog2.write_text(make_batch_log(50, 20)) + a, s, m = decide(dict(base, phase="phase2", worlds=50, days=20, log=str(plog2)), lambda pid, st, fr: False) check("phase2 PASS -> done", s["phase"] == "done" and m and "complete" in m) (td / "e.exit").write_text("0\n") tampered = dict(base, worlds=1, days=1) diff --git a/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj b/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj index b1b0ac2..a8aac24 100644 --- a/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj +++ b/src/LivingVillage.Desktop.Tests/LivingVillage.Desktop.Tests.fsproj @@ -10,6 +10,7 @@ <Compile Include="M6aTests.fs" /> <Compile Include="M6bTests.fs" /> <Compile Include="PrototypeTests.fs" /> + <Compile Include="M3BatchOutputTests.fs" /> <Compile Include="SampleTests.fs" /> </ItemGroup> diff --git a/src/LivingVillage.Desktop.Tests/M3BatchOutputTests.fs b/src/LivingVillage.Desktop.Tests/M3BatchOutputTests.fs new file mode 100644 index 0000000..e962e57 --- /dev/null +++ b/src/LivingVillage.Desktop.Tests/M3BatchOutputTests.fs @@ -0,0 +1,84 @@ +namespace LivingVillage.Desktop.Tests + +open System.Threading +open System.Threading.Tasks +open Microsoft.VisualStudio.TestTools.UnitTesting +open LivingVillage.Headless + +/// M3-1 可观测性回归：批次输出格式与逐世界流式驱动。 +[<TestClass>] +type M3BatchOutputTests () = + + let sampleBase = + "world=3 seed=45 chats_max=9 chats_avg=2.00 ratio=4.500 gini=0.100 relcnt_max=5 relcnt_min=1 relconn=6/6 nonfinite=0 check_a=PASS check_b=PASS check_c=PASS old_crit=PASS OK" + + [<TestMethod>] + member _.WorldLineKeepsPrefixAndAppendsElapsed () = + let line = BatchOutput.worldLine sampleBase 123.4 + Assert.IsTrue(line.StartsWith("world=3 seed=45 "), "world 行前缀必须保持不变") + Assert.IsTrue(line.StartsWith(sampleBase), "判据字段必须原样保留") + Assert.IsTrue(line.EndsWith("elapsed_s=123.4"), "耗时字段追加在行尾") + // 旧判据 token 仍在（监督器据此判定 OK）。 + Assert.AreEqual<string>(sampleBase + " elapsed_s=123.4", line) + + [<TestMethod>] + member _.DoneLineReportsProgressAndElapsed () = + Assert.AreEqual<string>("[done 2/4] elapsed=12.3s world=1", BatchOutput.doneLine 2 4 12.3 1) + Assert.AreEqual<string>("[done 4/4] elapsed=600.0s world=3", BatchOutput.doneLine 4 4 600.0 3) + + [<TestMethod>] + member _.SummaryLineKeepsOldFieldsAndAppendsWallClock () = + let line = BatchOutput.summaryLine 4 3 1 4 1.100 9.900 0.010 0.400 4 610.5 + // 监督器正则只读前三个字段，前缀必须逐字保持。 + Assert.IsTrue(line.StartsWith("batch_summary worlds=4 passed=3 failed=1 old_passed=4/4 "), line) + Assert.IsTrue(line.Contains("ratio_min=1.100 ratio_max=9.900 gini_min=0.010 gini_max=0.400"), line) + Assert.IsTrue(line.Contains("elapsed_s=610.5"), line) + Assert.IsTrue(line.EndsWith("workers=4 wall_s=610.5"), line) + + [<TestMethod>] + member _.CostLineDerivesThroughputFromMeasuredTicks () = + let line = BatchOutput.costLine 10L 51840000L 254.0 12345678L 3L 1L 0L + Assert.IsTrue(line.StartsWith("cost days=10 ticks=51840000 elapsed_s=254.000 "), line) + Assert.IsTrue(line.Contains("ticks_per_s=204094.5"), line) + Assert.IsTrue(line.EndsWith("allocated_bytes=12345678 gen0=3 gen1=1 gen2=0"), line) + + [<TestMethod>] + member _.DefaultCostLadderIsOneFiveTenTwentyFiftyHundred () = + CollectionAssert.AreEqual([| 1L; 5L; 10L; 20L; 50L; 100L |], List.toArray BatchOutput.defaultCostLadder) + + [<TestMethod>] + member _.RunStreamingEmitsEachWorldBeforeTheNextStarts () = + // workers=1：每次 work 后必须立刻 emit 该世界（证明不是收完再打印）。 + let events = ResizeArray<string> () + let results = + BatchOutput.runStreaming 3 1 + (fun k -> + lock events (fun () -> events.Add(sprintf "work%d" k)) + k * k) + (fun k _ _ -> + lock events (fun () -> events.Add(sprintf "emit%d" k))) + CollectionAssert.AreEqual([| 0; 1; 4 |], results) + CollectionAssert.AreEqual( + [| "work0"; "emit0"; "work1"; "emit1"; "work2"; "emit2" |], + events.ToArray()) + + [<TestMethod>] + member _.RunStreamingIsCompleteUnderParallelism () = + let total = 12 + let emitted = System.Collections.Concurrent.ConcurrentDictionary<int, int> () + let completions = ResizeArray<int> () + let results = + BatchOutput.runStreaming total 4 + (fun k -> k + 100) + (fun k result doneCount -> + Assert.AreEqual<int>(k + 100, result) + emitted.[k] <- doneCount + lock completions (fun () -> completions.Add doneCount)) + Assert.AreEqual<int>(total, results.Length) + Assert.AreEqual<int>(total, emitted.Count) + for k in 0 .. total - 1 do + Assert.IsTrue(emitted.ContainsKey k, sprintf "world=%d 未 emit" k) + // 完成计数恰为 1..total 的一个排列（并发下不重不漏）。 + CollectionAssert.AreEqual( + [| 1 .. total |], + completions |> Seq.sort |> Seq.toArray) diff --git a/src/LivingVillage.Headless/BatchOutput.fs b/src/LivingVillage.Headless/BatchOutput.fs new file mode 100644 index 0000000..aeaef52 --- /dev/null +++ b/src/LivingVillage.Headless/BatchOutput.fs @@ -0,0 +1,75 @@ +namespace LivingVillage.Headless + +open System.Threading.Tasks + +/// M3 批次的输出格式与流式并行驱动。 +/// +/// 全部是纯函数 / 无模拟依赖的驱动，便于单测；真实运行由 `Program.runBatch` +/// 传入 `evalWorld` 与 `Console.Out`。格式约定： +/// - `world=... ` 判据行保持原前缀与字段，行尾追加 `elapsed_s=<秒>`； +/// - 每完成一个世界追加一行 `[done k/N] elapsed=…s world=…`； +/// - `batch_summary` 保留旧字段，行尾追加 `workers=` 与 `wall_s=`（总墙钟）。 +module BatchOutput = + + /// `--cost-probe` 默认天数阶梯（单世界逐段实测）。 + let defaultCostLadder : int64 list = [ 1L; 5L; 10L; 20L; 50L; 100L ] + + /// world 结果行：前缀不变，行尾追加耗时字段。 + let worldLine (baseLine: string) (elapsedSeconds: float) : string = + sprintf "%s elapsed_s=%.1f" baseLine elapsedSeconds + + /// 逐世界完成进度行。 + let doneLine (completed: int) (total: int) (elapsedSeconds: float) (world: int) : string = + sprintf "[done %d/%d] elapsed=%.1fs world=%d" completed total elapsedSeconds world + + /// batch_summary：旧字段原样保留，追加 workers 与 wall_s（总墙钟，append 不删旧字段）。 + let summaryLine + (worlds: int) + (passed: int) + (failed: int) + (oldPassed: int) + (ratioMin: float) + (ratioMax: float) + (giniMin: float) + (giniMax: float) + (workers: int) + (elapsedSeconds: float) : string = + sprintf + "batch_summary worlds=%d passed=%d failed=%d old_passed=%d/%d ratio_min=%.3f ratio_max=%.3f gini_min=%.3f gini_max=%.3f elapsed_s=%.1f workers=%d wall_s=%.1f" + worlds passed failed oldPassed worlds ratioMin ratioMax giniMin giniMax elapsedSeconds workers elapsedSeconds + + /// --cost-probe 单段输出：ticks/s、墙钟、分配与 GC 全部来自真实测量。 + let costLine + (days: int64) + (ticks: int64) + (elapsedSeconds: float) + (allocatedBytes: int64) + (gen0: int64) + (gen1: int64) + (gen2: int64) : string = + let ticksPerSecond = if elapsedSeconds > 0.0 then float ticks / elapsedSeconds else 0.0 + sprintf + "cost days=%d ticks=%d elapsed_s=%.3f ticks_per_s=%.1f allocated_bytes=%d gen0=%d gen1=%d gen2=%d" + days ticks elapsedSeconds ticksPerSecond allocatedBytes gen0 gen1 gen2 + + /// 并行跑 total 个工作，**完成即回调** `emit world result completedCount`。 + /// 返回按 world 序号存放的结果数组；`work` 必须是仅依赖 k 的确定性纯世界计算。 + let runStreaming (total: int) (workers: int) (work: int -> 'T) (emit: int -> 'T -> int -> unit) : 'T[] = + let results = Array.zeroCreate total + let gate = obj () + let mutable completed = 0 + let options = ParallelOptions(MaxDegreeOfParallelism = max 1 workers) + Parallel.For( + 0, + total, + options, + fun k -> + let result = work k + let completedCount = + lock gate (fun () -> + results.[k] <- result + completed <- completed + 1 + completed) + emit k result completedCount) + |> ignore + results diff --git a/src/LivingVillage.Headless/LivingVillage.Headless.fsproj b/src/LivingVillage.Headless/LivingVillage.Headless.fsproj index d9cd0ab..bf8fd7b 100644 --- a/src/LivingVillage.Headless/LivingVillage.Headless.fsproj +++ b/src/LivingVillage.Headless/LivingVillage.Headless.fsproj @@ -7,6 +7,7 @@ <ItemGroup> <Compile Include="PerformanceProbe.fs" /> + <Compile Include="BatchOutput.fs" /> <Compile Include="Program.fs" /> </ItemGroup> diff --git a/src/LivingVillage.Headless/Program.fs b/src/LivingVillage.Headless/Program.fs index 2a2482b..89f4f26 100644 --- a/src/LivingVillage.Headless/Program.fs +++ b/src/LivingVillage.Headless/Program.fs @@ -449,38 +449,75 @@ let runBatch (worlds: int) (days: int64) : int = // 明星集中型分布过严，relcnt_max=0 即判死，与"非均匀分布"的验收目标不符） // (c) 无 NaN/Inf printfn "checks: (a) chats max/avg > 2.0 (b) npcs_with_relation >= 0.6*npc_count (c) no NaN (old) relcnt_max > 2*relcnt_min (report only)" - // 世界间相互独立（seed=42+k，逐世界确定性），可安全并行；输出仍按 world 序号 - // 顺序打印，逐行内容与串行版本逐字节一致。LV_BATCH_WORKERS 可覆盖（默认 modest 4）。 + // 世界间相互独立（seed=42+k，逐世界确定性），可安全并行；每个世界一完成即流式 + // 打印该行与进度（完成序，可能非 world 升序），判据行内容与串行版本逐字节一致。 + // LV_BATCH_WORKERS 可覆盖（默认 modest 4）。 let workers = match Environment.GetEnvironmentVariable "LV_BATCH_WORKERS" with | null | "" -> min 4 Environment.ProcessorCount | v -> match Int32.TryParse v with true, n when n > 0 -> min n Environment.ProcessorCount | _ -> min 4 Environment.ProcessorCount printfn "workers=%d" workers - let results = Array.zeroCreate worlds - System.Threading.Tasks.Parallel.For - (0, worlds, - System.Threading.Tasks.ParallelOptions(MaxDegreeOfParallelism = workers), - fun k -> results.[k] <- evalWorld days k) - |> ignore + Console.Out.Flush() + let wall = System.Diagnostics.Stopwatch.StartNew() + // 逐世界流式：每个世界一完成立刻打印该行（行尾追加 elapsed_s），再打印一行 + // [done k/N] 进度，并 flush，使长批次不再整轮零进度。结果仍按 world 序号落数组。 + let results = + LivingVillage.Headless.BatchOutput.runStreaming worlds workers + (fun k -> + let sw = System.Diagnostics.Stopwatch.StartNew() + let verdict = evalWorld days k + sw.Stop() + (verdict, sw.Elapsed.TotalSeconds)) + (fun k (verdict, seconds) doneCount -> + printfn "%s" (LivingVillage.Headless.BatchOutput.worldLine verdict.Line seconds) + printfn "%s" (LivingVillage.Headless.BatchOutput.doneLine doneCount worlds seconds k) + Console.Out.Flush()) + wall.Stop() let mutable passed = 0 let mutable failed = 0 let mutable oldPassed = 0 let ratios = ResizeArray<float> () let ginis = ResizeArray<float> () - let sw = System.Diagnostics.Stopwatch.StartNew() - for r in results do - printfn "%s" r.Line - if r.Ok then passed <- passed + 1 else failed <- failed + 1 - if r.CheckOld then oldPassed <- oldPassed + 1 - ratios.Add r.Ratio - ginis.Add r.Gini - sw.Stop() + for verdict, _ in results do + if verdict.Ok then passed <- passed + 1 else failed <- failed + 1 + if verdict.CheckOld then oldPassed <- oldPassed + 1 + ratios.Add verdict.Ratio + ginis.Add verdict.Gini let allPass = failed = 0 - printfn "batch_summary worlds=%d passed=%d failed=%d old_passed=%d/%d ratio_min=%.3f ratio_max=%.3f gini_min=%.3f gini_max=%.3f elapsed_s=%.1f" - worlds passed failed oldPassed worlds (Seq.min ratios) (Seq.max ratios) (Seq.min ginis) (Seq.max ginis) sw.Elapsed.TotalSeconds + printfn "%s" + (LivingVillage.Headless.BatchOutput.summaryLine + worlds passed failed oldPassed (Seq.min ratios) (Seq.max ratios) (Seq.min ginis) (Seq.max ginis) workers wall.Elapsed.TotalSeconds) printfn "M3_ACCEPTANCE=%s" (if allPass then "PASS" else "FAIL") + Console.Out.Flush() if allPass then 0 else 1 +/// `--cost-probe`：单世界按天数阶梯逐段真实测量（墙钟、ticks/s、分配与 GC）， +/// 只读、不改任何模拟语义。天数阶梯可用 `--cost-probe D1,D2,...` 覆盖。 +let runCostProbe (ladder: int64 list) : int = + printfn "cost_probe ladder=%s npcs=%d" (ladder |> List.map string |> String.concat ",") Sim.npcCount + Console.Out.Flush() + for days in ladder do + let ticks = days * ticksPerDay + System.GC.Collect() + let allocBefore = System.GC.GetTotalAllocatedBytes(false) + let gen0Before = System.GC.CollectionCount 0 + let gen1Before = System.GC.CollectionCount 1 + let gen2Before = System.GC.CollectionCount 2 + let sw = System.Diagnostics.Stopwatch.StartNew() + let stats = runSimulation false days 42UL Sim.npcCount + sw.Stop() + let allocated = System.GC.GetTotalAllocatedBytes(false) - allocBefore + let line = + LivingVillage.Headless.BatchOutput.costLine + days ticks sw.Elapsed.TotalSeconds allocated + (int64 (System.GC.CollectionCount 0 - gen0Before)) + (int64 (System.GC.CollectionCount 1 - gen1Before)) + (int64 (System.GC.CollectionCount 2 - gen2Before)) + printfn "%s" line + printfn "cost_probe_check days=%d nonfinite=%d out_of_bounds=%d final_tick=%d" days stats.NonFinite stats.OutOfBounds stats.World.Tick + Console.Out.Flush() + 0 + let runExisting (argv: string[]) = let rec parse (i: int) @@ -544,14 +581,32 @@ let runExisting (argv: string[]) = if stats.NonFinite > 0L || stats.OutOfBounds > 0L || not replayOk then 1 else 0 | Error msg -> eprintfn $"headless: {msg}" - eprintfn "usage: dotnet run -c Release --project src/LivingVillage.Headless -- --days N --seed S [--npc N] [--dump-relations] [--dump-rumors] [--replay-rumors] | --m5-smoke | --m6a-smoke | --batch K D | --performance-baseline" + eprintfn "usage: dotnet run -c Release --project src/LivingVillage.Headless -- --days N --seed S [--npc N] [--dump-relations] [--dump-rumors] [--replay-rumors] | --m5-smoke | --m6a-smoke | --batch K D | --performance-baseline | --cost-probe [D1,D2,...]" 2 +let private parseCostLadder (argv: string[]) : int64 list = + if argv.Length >= 2 then + let parsed = + argv.[1].Split(',') + |> Array.toList + |> List.map (fun token -> + match Int64.TryParse(token.Trim()) with + | true, days when days > 0L -> Some days + | _ -> None) + if parsed <> [] && List.forall Option.isSome parsed then + parsed |> List.map Option.get + else + LivingVillage.Headless.BatchOutput.defaultCostLadder + else + LivingVillage.Headless.BatchOutput.defaultCostLadder + [<EntryPoint>] let main argv = if argv.Length = 1 && argv.[0] = "--m6a-smoke" then runM6aSmoke () elif argv.Length = 1 && argv.[0] = "--performance-baseline" then LivingVillage.Headless.PerformanceProbe.runDefault () + elif argv.Length >= 1 && argv.[0] = "--cost-probe" then + runCostProbe (parseCostLadder argv) else runExisting argv |