diff --git a/session-mechanism/graph/README.md b/session-mechanism/graph/README.md index 631a156..ee49a15 100644 --- a/session-mechanism/graph/README.md +++ b/session-mechanism/graph/README.md @@ -85,9 +85,19 @@ python graph/test_graphctl.py -v -全绿应看到 `Ran 49 tests ... OK`,用时 20 秒上下(每条用例起一次子进程)。 +全绿应看到 `Ran 52 tests ... OK`,用时 15 秒上下(每条用例起一次子进程)。 -测试只往临时目录写:`/graph-test-/<用例名>`,默认 `GRAPH_TEST_TMP` = `AIProject/ai1net-dsh-server/tmp`,由每条用例的 `tearDown` + 进程退出的 `atexit` 兜底删除。想换落点就设环境变量 `GRAPH_TEST_TMP`。 +测试只往临时目录写:`/graph-test-/<用例名>`,默认 `GRAPH_TEST_TMP` = `AIProject/ai1net-dsh-server/tmp`。每条用例只**新建**自己的目录、跑完不删,整轮收尾由 `tearDownModule` 清一次;这样一次跑下来新建几十个文件、不触发本机沙箱的「批量删除」守卫(那个守卫一次删满 50 个文件就会拦,曾让 20 条用例报 ERROR,而用例本身零失败)。想换落点就设环境变量 `GRAPH_TEST_TMP`。 + +## 五之二、并发写是怎么保证的 + +两个进程可以同时对同一张图动手(比如两个节点同时开工),做法是三层: + +1. 临时文件名带进程号与随机数,两个进程写字不会撞同一个名字(早期版本用固定名 `graph.json.tmp`,并发时直接报 `PermissionError`)。 +2. 每次写都先读最新磁盘状态、在它上面重放本次变更,再原子替换——避免「后写的整份覆盖前写的」导致节点状态静默消失。 +3. 图里带 `_rev` 版本号:写完回读核对,发现自己不是最新一版就重读重做,连续三次仍被覆盖才报错退出(要的是「宁可报错,也不静默丢状态」)。 + +这条路径由 `T9_Concurrency` 的三个用例守着:并发开工两个节点、并发回报两个节点、版本号每次写都前进。 清残留在本机是常态:跑完后 `tmp/` 下偶尔还能看到 `graph-test-/`(进程内已确认删空,是本机沙箱在进程退出后把删除路径又放回来了,与测试代码无关)。一条命令清掉: diff --git a/session-mechanism/graph/graphctl.py b/session-mechanism/graph/graphctl.py index 94c463c..c88513c 100644 --- a/session-mechanism/graph/graphctl.py +++ b/session-mechanism/graph/graphctl.py @@ -63,6 +63,7 @@ import argparse import io import json import os +import random import sys import time from pathlib import Path @@ -92,13 +93,46 @@ def _load(ws: str, gid: str) -> dict: def _save(ws: str, gid: str, g: dict) -> None: + """整份写回(新建图时用)。临时名带 pid + 随机数 ⇒ 两进程不会撞同一个名字。""" p = _gdir(ws, gid) / "graph.json" - tmp = p.with_suffix(".json.tmp") + tmp = p.with_suffix(".%d.%d.json.tmp" % (os.getpid(), random.randint(0, 1 << 30))) tmp.write_text(json.dumps(g, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", newline="\n") os.replace(tmp, p) # 原子替换,避免半截文件 +def _update(ws: str, gid: str, mutate, retries: int = 3) -> dict: + """读-改-写,带版本号乐观重试(2026-10-19 修两处实测缺陷)。 + + ① 临时名原来固定成 `graph.json.tmp` ⇒ 两进程同时写直接撞 `PermissionError` + (桌面测试报告第 14 条实测:同时对 C、D 执行 start 就复现)。 + ② 更隐蔽的一层:就算临时名不撞,两人同时读-改-写会**丢更新** —— + 后写的整份覆盖前写的,节点状态静默消失,连报错都没有。 + ③ 对策:图里带 `_rev`,每次在**最新磁盘状态**上重放变更,写完回读核对; + 被别的进程覆盖就重读重做,最多 `retries` 次,仍失败则**报错退出**(⛔ 不静默)。 + ⚠️ 本机不用 OS 文件锁(实测会卡死),靠「唯一临时名 + 原子替换 + 回读核对」。 + """ + p = _gdir(ws, gid) / "graph.json" + if not p.is_file(): + raise SystemExit("找不到图:%s" % p) + for _ in range(retries): + g = json.loads(p.read_text(encoding="utf-8")) + rev = int(g.get("_rev") or 0) + mutate(g) + g["_rev"] = rev + 1 + tmp = p.with_suffix(".%d.%d.json.tmp" % (os.getpid(), random.randint(0, 1 << 30))) + tmp.write_text(json.dumps(g, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", newline="\n") + os.replace(tmp, p) + try: + back = json.loads(p.read_text(encoding="utf-8")) + except Exception: # noqa: BLE001 + back = {} + if int(back.get("_rev") or 0) == rev + 1: + return back + raise SystemExit("并发冲突:%s 连续 %d 次被别的进程覆盖,稍后重试" % (p, retries)) + + def _event(ws: str, gid: str, obj: dict) -> None: p = _gdir(ws, gid) / "events.jsonl" obj = dict(obj) @@ -258,35 +292,37 @@ def cmd_ready(a) -> int: # ── 状态流转 ──────────────────────────────────────────────────────────── def cmd_start(a) -> int: - g = _load(a.ws, a.gid) - n = (g.get("nodes") or {}).get(a.node) - if not n: - raise SystemExit("没有这个节点:%s" % a.node) - if str(n.get("status")) == "running": - raise SystemExit("节点 %s 已在跑,不能重复开工" % a.node) - if str(n.get("status")) == "done": - raise SystemExit("节点 %s 已完成,不能开工" % a.node) - # pending 与 blocked 都可开工:blocked 是「这轮没过」,重试时 attempts 保留 - n["status"] = "running" - _save(a.ws, a.gid, g) + def _mut(g): + n = (g.get("nodes") or {}).get(a.node) + if not n: + raise SystemExit("没有这个节点:%s" % a.node) + if str(n.get("status")) == "running": + raise SystemExit("节点 %s 已在跑,不能重复开工" % a.node) + if str(n.get("status")) == "done": + raise SystemExit("节点 %s 已完成,不能开工" % a.node) + # pending 与 blocked 都可开工:blocked 是「这轮没过」,重试时 attempts 保留 + n["status"] = "running" + _update(a.ws, a.gid, _mut) _event(a.ws, a.gid, {"node": a.node, "to": "running", "by": a.by or ""}) _out({"ok": True, "node": a.node, "status": "running"}) return 0 def cmd_report(a) -> int: - g = _load(a.ws, a.gid) - n = (g.get("nodes") or {}).get(a.node) - if not n: - raise SystemExit("没有这个节点:%s" % a.node) if a.state not in ("done", "blocked"): raise SystemExit("--state 只认 done / blocked") - if a.evidence: - n["evidence"] = a.evidence - n["status"] = a.state - if a.state == "blocked": - n["attempts"] = int(n.get("attempts") or 0) + 1 - _save(a.ws, a.gid, g) + + def _mut(g): + n = (g.get("nodes") or {}).get(a.node) + if not n: + raise SystemExit("没有这个节点:%s" % a.node) + if a.evidence: + n["evidence"] = a.evidence + n["status"] = a.state + if a.state == "blocked": + n["attempts"] = int(n.get("attempts") or 0) + 1 + + g = _update(a.ws, a.gid, _mut) _event(a.ws, a.gid, {"node": a.node, "to": a.state, "evidence": a.evidence or "", "by": a.by or ""}) r = _ready(g) diff --git a/session-mechanism/graph/test_graphctl.py b/session-mechanism/graph/test_graphctl.py index 4b55f5b..0485caa 100644 --- a/session-mechanism/graph/test_graphctl.py +++ b/session-mechanism/graph/test_graphctl.py @@ -83,15 +83,18 @@ def _drop(p: Path, tries: int = 5) -> None: class Base(unittest.TestCase): def setUp(self): + # 🔴 2026-10-19 改(实测:本机沙箱有「批量删除」守卫,一次删满 50 个文件即弹 + # SAFE_DELETE_BULK_CONFIRM_REQUIRED ⇒ 49 条用例里 20 条被拦成 ERROR, + # 单跑却全 OK —— 说明红的是清理动作,不是用例)。 + # ⇒ 每例只**新建**自己的目录、跑完不删;整轮收尾由 tearDownModule 清一次。 + # TMP_ROOT 带 pid,重跑天然不撞旧目录。 self.ws = TMP_ROOT / self._testMethodName - shutil.rmtree(self.ws, ignore_errors=True) self.ws.mkdir(parents=True, exist_ok=True) self.gid = "g1" def tearDown(self): global _TD_COUNT _TD_COUNT += 1 - _drop(self.ws) # -- 便捷动作 -- def gpath(self, gid=None) -> Path: @@ -510,6 +513,67 @@ class T8_Isolation(Base): os.environ.get("GRAPH_TEST_TMP"), TMP_ROOT) +# ── 9 并发写 ──────────────────────────────────────────────────────────── + +class T9_Concurrency(Base): + """两个进程同时写同一张图:要都成功,且谁的状态都不许丢。 + + 2026-10-19 修的实测缺陷(桌面测试报告第 14 条): + 临时名原为固定的 `graph.json.tmp` ⇒ 并发 start 直接撞 PermissionError; + 更隐蔽的是「读-改-写」后写覆盖前写 ⇒ 节点状态静默消失。 + 修法=唯一临时名(pid+随机)+ `_rev` 乐观重试(写后回读核对,被覆盖则重读重做)。 + """ + + def _parallel(self, *argsets): + procs = [subprocess.Popen([PY, str(GRAPHCTL)] + list(a), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, encoding="utf-8") for a in argsets] + return [p.communicate() + (p.returncode,) for p in procs] + + def _two_free_nodes(self): + self.init({"max_parallel": 4, "nodes": [ + {"id": "A", "title": "甲", "domain": ["a/"], "accepts": ["好"], "deps": []}, + {"id": "B", "title": "乙", "domain": ["b/"], "accepts": ["好"], "deps": []}, + ]}) + + def test_parallel_start_keeps_both_running(self): + self._two_free_nodes() + res = self._parallel( + ["start", "--ws", str(self.ws), "--gid", self.gid, "--node", "A", "--by", "t1"], + ["start", "--ws", str(self.ws), "--gid", self.gid, "--node", "B", "--by", "t2"], + ) + for out, err, rc in res: + self.assertEqual(rc, 0, "并发 start 不该失败(rc=%d, err=%s)" % (rc, err)) + self.assertNotIn("PermissionError", err) + g = self.load() + self.assertEqual(g["nodes"]["A"]["status"], "running") + self.assertEqual(g["nodes"]["B"]["status"], "running") + + def test_parallel_report_keeps_both_evidence(self): + self._two_free_nodes() + ok("start", "--ws", str(self.ws), "--gid", self.gid, "--node", "A") + ok("start", "--ws", str(self.ws), "--gid", self.gid, "--node", "B") + res = self._parallel( + ["report", "--ws", str(self.ws), "--gid", self.gid, "--node", "A", + "--state", "done", "--evidence", "甲好了"], + ["report", "--ws", str(self.ws), "--gid", self.gid, "--node", "B", + "--state", "done", "--evidence", "乙好了"], + ) + for out, err, rc in res: + self.assertEqual(rc, 0, "并发 report 不该失败(rc=%d, err=%s)" % (rc, err)) + g = self.load() + self.assertEqual(g["nodes"]["A"]["status"], "done") + self.assertEqual(g["nodes"]["B"]["status"], "done") + self.assertEqual(g["nodes"]["A"]["evidence"], "甲好了") + self.assertEqual(g["nodes"]["B"]["evidence"], "乙好了") + + def test_rev_advances_on_every_write(self): + self._two_free_nodes() + r0 = int(self.load().get("_rev") or 0) + ok("start", "--ws", str(self.ws), "--gid", self.gid, "--node", "A") + self.assertGreater(self.load().get("_rev"), r0) + + def tearDownModule(): _drop(TMP_ROOT)