From 89cd13517446d44c3cc57fc9e7dccccfa54874e4 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 12:14:39 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=A4=9A=E8=BF=9B=E7=A8=8B=E5=8D=8F?= =?UTF-8?q?=E8=B0=83=E7=9A=84=E5=8D=95=E6=B5=8B=EF=BC=88=E4=B8=A4=E7=A7=8D?= =?UTF-8?q?=E5=BA=93=EF=BC=89=EF=BC=9A=E5=88=AB=E7=9A=84=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E7=94=A8=E5=BA=93=E9=87=8C=E7=9A=84=E8=A1=8C=E8=A3=85=E5=87=BA?= =?UTF-8?q?=E6=9D=A5=EF=BC=88=E7=99=BB=E8=AE=B0=E7=9D=80=E7=9A=84=E5=90=8C?= =?UTF-8?q?=E4=BC=B4=E3=80=81=E4=BB=8E=E6=B2=A1=E7=99=BB=E8=AE=B0=E8=BF=87?= =?UTF-8?q?=E7=9A=84=E5=B7=B2=E7=BB=8F=E6=B2=A1=E4=BA=86=E7=9A=84=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=EF=BC=89=E2=80=94=E2=80=94=E8=B0=81=E8=BF=98=E6=B4=BB?= =?UTF-8?q?=E7=9D=80=E3=80=81=E5=90=8D=E9=A2=9D=E6=8C=89=E4=BC=98=E5=85=88?= =?UTF-8?q?=E7=BA=A7=E5=85=88=E7=BB=99=E4=BA=A4=E4=BA=92=E7=B1=BB=E3=80=81?= =?UTF-8?q?=E4=B8=80=E7=B1=BB=E5=88=B0=E4=BA=86=E4=B8=8A=E9=99=90=E4=B8=8D?= =?UTF-8?q?=E6=8C=A1=E5=88=AB=E7=9A=84=E7=B1=BB=E3=80=81=E6=B2=A1=E4=BA=86?= =?UTF-8?q?=E7=9A=84=E8=BF=9B=E7=A8=8B=E5=8D=A0=E7=9D=80=E7=9A=84=E5=90=8D?= =?UTF-8?q?=E9=A2=9D=E6=94=B6=E5=9B=9E=E8=80=8C=E6=B4=BB=E7=9D=80=E7=9A=84?= =?UTF-8?q?=E5=90=8C=E4=BC=B4=E7=9A=84=E4=B8=8D=E7=A2=B0=E3=80=81=E5=90=8C?= =?UTF-8?q?=E6=BA=90=E9=98=9F=E5=87=A0=E4=B8=AA=E8=BF=9B=E7=A8=8B=E5=85=B1?= =?UTF-8?q?=E7=94=A8=E4=B8=94=E8=B7=B3=E8=BF=87=E6=B2=A1=E4=BA=86=E7=9A=84?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E7=9A=84=E4=BD=8D=E7=BD=AE=E3=80=81=E6=8E=A5?= =?UTF-8?q?=E6=89=8B=E8=BF=90=E8=A1=8C=E5=8F=AA=E7=A2=B0=E5=B1=9E=E4=B8=BB?= =?UTF-8?q?=E4=B8=8D=E5=9C=A8=E4=BA=86=E7=9A=84=EF=BC=88=E5=8D=87=E7=BA=A7?= =?UTF-8?q?=E5=89=8D=E6=B2=A1=E8=AE=B0=E5=B1=9E=E4=B8=BB=E7=9A=84=E7=85=A7?= =?UTF-8?q?=E4=B8=8D=E5=9C=A8=E4=BA=86=E5=A4=84=E7=90=86=EF=BC=89=E3=80=81?= =?UTF-8?q?=E5=81=9C=E6=AD=A2=E5=92=8C=E8=AE=B0=E4=BA=8B=E4=BB=B6=E4=BA=A4?= =?UTF-8?q?=E7=BB=99=E6=B4=BB=E7=9D=80=E7=9A=84=E5=B1=9E=E4=B8=BB=E3=80=81?= =?UTF-8?q?=E5=B1=9E=E4=B8=BB=E4=B8=8D=E5=9C=A8=E4=BA=86=E7=9A=84=E5=81=9C?= =?UTF-8?q?=E6=AD=A2=E7=9B=B4=E6=8E=A5=E6=8E=A5=E6=89=8B=E6=94=BE=E5=BC=83?= =?UTF-8?q?=E3=80=81=E5=BD=93=E7=8F=AD=E5=88=B0=E6=9C=9F=E4=BA=A4=E6=8E=A5?= =?UTF-8?q?=E3=80=81=E5=90=8C=E4=B8=80=E6=9C=AC=E6=AF=8F=E6=97=A5=E8=B4=A6?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E8=AE=B0=E4=B8=8D=E8=B6=8A=E8=BF=87=E4=B8=8A?= =?UTF-8?q?=E9=99=90=E3=80=81SQLite=20=E7=AC=AC=E4=BA=8C=E4=B8=AA=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=EF=BC=88=E4=BB=A5=E5=8F=8A=20AGENTLAB=5FWORKERS=20?= =?UTF-8?q?=E5=A4=A7=E4=BA=8E=201=EF=BC=89=E6=8B=92=E7=BB=9D=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=EF=BC=9B=E8=AF=84=E6=B5=8B=E8=AF=B4=E6=98=8E=E9=87=8C?= =?UTF-8?q?=E7=9A=84=E5=B9=B6=E5=8F=91=E4=B8=80=E6=AE=B5=E7=85=A7=E7=8E=B0?= =?UTF-8?q?=E5=9C=A8=E7=9A=84=E5=90=8D=E9=A2=9D=E8=AF=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/eval/runner.py | 5 +- backend/tests/test_cluster.py | 287 ++++++++++++++++++++++++++++++++++ 2 files changed, 290 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_cluster.py diff --git a/backend/app/eval/runner.py b/backend/app/eval/runner.py index 3f179db4..a7bb8ecf 100644 --- a/backend/app/eval/runner.py +++ b/backend/app/eval/runner.py @@ -12,8 +12,9 @@ EvalResult：建评测时就全部建好（pending），进度、「正在做什 snapshot 视图）：起草、剖析照常进行，也不影响这次评测的可比性。各数据源的上传表格版本、表结构哈希开始和结束各记一次， 变了在指纹里标出（sources_changed）。 -**并发。** 同时最多 max_parallel 题，且所有臂合起来不超过全局运行并发的四分之一（全局只有一个信号量、没有优先级，评测不能 -挤占真实用户的运行）。A/B 的同一题几臂同时开跑。每题时限 per_case_timeout_s：作答一臂超时就取消它正在跑的运行、记超时； +**并发。** 同时最多 max_parallel 题，且所有臂合起来不超过全局运行并发的四分之一。评测发起的作答、算期望的运行归「系统」类 +（app/cluster/slots.py）：全站的名额按类别有上限和优先级，评测再多也只占得到系统那一类的名额，问数据、试跑优先，不被挤占。 +A/B 的同一题几臂同时开跑。每题时限 per_case_timeout_s：作答一臂超时就取消它正在跑的运行、记超时； 期望探针每次也按这个时限。 **预算**（评测-D9；单独一本账 engine/spend.eval_spend，不挤占裁判、核查的额度）：开跑前按上一次的每题平均花费估算（estimate）。 diff --git a/backend/tests/test_cluster.py b/backend/tests/test_cluster.py new file mode 100644 index 00000000..e0f6f34d --- /dev/null +++ b/backend/tests/test_cluster.py @@ -0,0 +1,287 @@ +"""多进程协调（app/cluster，docs/design/orchestration-round §5 W2-b）在一个进程里能验的那几样：别的进程用库里的行装出来 +（一个登记着、租约没过期的「同伴」，一个从没登记过的「已经没了的进程」）。两种库都跑；真起两个进程的端到端见 +test_multiprocess_e2e.py。""" +from __future__ import annotations + +import asyncio +import os +import subprocess +import sys +from datetime import timedelta +from pathlib import Path + +import pytest +from sqlalchemy import delete, select + +from app.cluster import identity, leader as leader_mod, slots, state, workers +from app.db.base import SessionLocal, new_id, utcnow +from app.db.models import Run, RunEvent +from app.db.models_cluster import ClusterLease, ClusterRequest, ClusterWorker, LaneTicket, RunSlot + +PEER = "peer-host:4242:feedbeef" +GHOST = "gone-host:999:deadbeef" + + +@pytest.fixture(autouse=True) +async def clean(): + async def wipe() -> None: + async with SessionLocal() as session: + for model in (RunSlot, LaneTicket, ClusterRequest, ClusterWorker): + await session.execute(delete(model)) + await session.execute(delete(ClusterLease).where(ClusterLease.name.like("test-%"))) + await session.commit() + state.forget() + slots.forget_policy() + + await wipe() + yield + await wipe() + + +async def _peer(owner: str = PEER, *, alive: bool = True) -> None: + now = utcnow() + async with SessionLocal() as session: + session.add(ClusterWorker(id=owner, host="peer-host", pid=4242, started_at=now, heartbeat_at=now, + lease_until=now + timedelta(seconds=60 if alive else -60))) + await session.commit() + state.forget() + + +def _policy(monkeypatch, total: int, **classes: tuple[int, int]) -> None: + value = slots.values({"max_total": total, "classes": {k: {"limit": lim, "priority": pri} + for k, (lim, pri) in classes.items()}}) + + async def fixed(*, fresh: bool = False): # noqa: ARG001 + return value + + monkeypatch.setattr(slots, "policy", fixed) + + +async def test_liveness_counts_only_unexpired_peers_and_never_lists_itself(): + await workers.register() + await _peer(PEER, alive=True) + await _peer(GHOST, alive=False) + live = await workers.liveness() + assert live.others == frozenset({PEER}) + assert live.alive(PEER) and not live.alive(GHOST) and not live.alive(None) + assert not live.alive(identity.owner()), "属主是自己的只看本进程里是不是真的在跑" + assert live.alive(identity.owner(), local=True) and live.trusting().alive(identity.owner()) + + +async def test_slots_go_to_the_higher_priority_class_first(monkeypatch): + """全站 1 个名额：系统类的先占着、又一个系统类排着，后来的交互类先拿到空出来的名额；已经在跑的不打断。""" + _policy(monkeypatch, 1, interactive=(1, 40), system=(1, 10)) + order: list[str] = [] + release = {name: asyncio.Event() for name in ("e1", "e2", "chat")} + + async def run(name: str, origin: str) -> None: + async with slots.hold(f"run-{name}", origin): + order.append(name) + await release[name].wait() + + first = asyncio.create_task(run("e1", "eval")) + await asyncio.sleep(0.2) + second = asyncio.create_task(run("e2", "eval")) + await asyncio.sleep(0.2) + chat = asyncio.create_task(run("chat", "chat")) + await asyncio.sleep(0.2) + assert order == ["e1"], "名额满了，后来的两个都在等" + release["e1"].set() + await asyncio.wait_for(asyncio.shield(_until(lambda: len(order) == 2)), timeout=10) + assert order == ["e1", "chat"], "交互类的优先级高，先拿到名额" + release["chat"].set() + release["e2"].set() + await asyncio.wait_for(asyncio.gather(first, second, chat), timeout=10) + assert order == ["e1", "chat", "e2"] + + +async def test_a_class_at_its_cap_does_not_hold_up_the_others(monkeypatch): + """系统类的上限是 1：第二个系统类的等着，交互类的照样拿到名额（全站还有空）。""" + _policy(monkeypatch, 5, interactive=(5, 40), system=(1, 10)) + got: list[str] = [] + gate = asyncio.Event() + + async def run(name: str, origin: str) -> None: + async with slots.hold(f"run-{name}", origin): + got.append(name) + await gate.wait() + + tasks = [asyncio.create_task(run("e1", "eval")), asyncio.create_task(run("e2", "recall"))] + await asyncio.sleep(0.3) + tasks.append(asyncio.create_task(run("chat", "canvas"))) + await _until(lambda: "chat" in got) + assert len({"e1", "e2"} & set(got)) == 1, "系统类到了上限：另一个等着" + snapshot = await slots.snapshot() + assert snapshot["classes"]["system"] == {"running": 1, "waiting": 1} + gate.set() + await asyncio.wait_for(asyncio.gather(*tasks), timeout=10) + + +async def test_slots_held_by_a_process_that_is_gone_are_taken_back(monkeypatch): + _policy(monkeypatch, 1, interactive=(1, 40)) + async with SessionLocal() as session: + session.add(RunSlot(run_id="ghost-run", klass="interactive", priority=40, owner=GHOST, granted=True)) + await session.commit() + async with slots.hold("mine", "chat"): + async with SessionLocal() as session: + rows = (await session.execute(select(RunSlot.run_id))).scalars().all() + assert rows == ["mine"], "没了的进程占着的名额收回了" + + +async def test_slots_of_a_live_peer_are_left_alone(monkeypatch): + _policy(monkeypatch, 1, interactive=(1, 40)) + await _peer(PEER) + async with SessionLocal() as session: + session.add(RunSlot(run_id="peer-run", klass="interactive", priority=40, owner=PEER, granted=True)) + await session.commit() + waiting = asyncio.create_task(_hold_once("mine")) + await asyncio.sleep(0.5) + assert not waiting.done(), "活着的同伴占着唯一的名额：等着" + async with SessionLocal() as session: + await session.execute(delete(RunSlot).where(RunSlot.run_id == "peer-run")) + await session.commit() + slots.wake_local() + await asyncio.wait_for(waiting, timeout=10) + + +async def _hold_once(run_id: str) -> None: + async with slots.hold(run_id, "chat"): + pass + + +async def test_the_source_lane_is_shared_and_skips_tickets_of_a_gone_process(): + from app.core import job_runtime as runtime + + import app.data.catalog_jobs # noqa: F401 - 登记 draft、profile + + async with SessionLocal() as session: + session.add(LaneTicket(source_id="src-x", kind="profile", job_id="ghost-job", owner=GHOST, granted=True)) + session.add(LaneTicket(source_id="src-y", kind="profile", job_id="peer-job", owner=PEER, granted=True)) + await session.commit() + await _peer(PEER) + mine = await runtime.lanes.enter("src-x", "draft", "job-1") + assert mine.granted.is_set(), "没了的进程占着的位置让出来了" + other = await runtime.lanes.enter("src-y", "draft", "job-2") + assert not other.granted.is_set(), "活着的同伴在这个源上做着事：排在后面" + assert await runtime.lanes.ahead("src-y", "job-2") == ["profile"] + with pytest.raises(runtime.LaneBusy) as busy: + await runtime.lanes.try_enter("src-y", "import") + assert busy.value.holder.kind == "profile" + await mine.release() + await other.release() + + +async def test_reaping_runs_only_touches_the_ones_whose_owner_is_gone(): + from app.engine.runner import run_manager + + await _peer(PEER) + async with SessionLocal() as session: + live_peer = Run(status="running", owner=PEER, graph={}, input={}) + dead = Run(status="running", owner=GHOST, graph={}, input={}) + queued = Run(status="queued", owner=GHOST, graph={}, input={}) + legacy = Run(status="running", owner=None, graph={}, input={}) + session.add_all([live_peer, dead, queued, legacy]) + await session.commit() + done = await run_manager.reap(await state.current(fresh=True)) + assert done >= 3 + async with SessionLocal() as session: + rows = {r.id: r for r in (await session.execute(select(Run).where( + Run.id.in_([live_peer.id, dead.id, queued.id, legacy.id])))).scalars()} + events = list((await session.execute(select(RunEvent.run_id, RunEvent.type, RunEvent.data).where( + RunEvent.run_id.in_([dead.id, queued.id])))).all()) + assert rows[live_peer.id].status == "running" and rows[live_peer.id].owner == PEER, "活着的同伴在跑的不碰" + assert rows[dead.id].status == "interrupted" and rows[dead.id].owner is None + assert rows[queued.id].status == "failed" + assert rows[legacy.id].status == "interrupted", "升级前没记属主的照「属主不在了」处理" + assert any(e.type == "log" and (e.data or {}).get("code") == "server_shutdown" for e in events) + assert any(e.type == "run.failed" for e in events) + + +async def test_stopping_or_noting_a_run_on_a_live_peer_goes_to_that_peer(): + from app.core.events import EventType + from app.engine.runner import run_manager + + await _peer(PEER) + async with SessionLocal() as session: + run = Run(status="running", owner=PEER, graph={}, input={}) + session.add(run) + await session.commit() + assert await run_manager.cancel(run.id, actor="张工") == "stopping" + await run_manager.note(run.id, EventType.LOG, level="info", message="别的进程记的一条") + async with SessionLocal() as session: + sent = list((await session.execute(select(ClusterRequest).where(ClusterRequest.target_id == run.id) + .order_by(ClusterRequest.created_at))).scalars()) + row = await session.get(Run, run.id) + assert [(r.owner, r.kind, r.target) for r in sent] == [(PEER, "stop", "run"), (PEER, "note", "run")] + assert sent[0].actor == "张工" and sent[1].payload["data"]["message"] == "别的进程记的一条" + assert row.status == "running", "停止交给属主去做，这里不改状态" + + +async def test_stopping_a_run_whose_owner_is_gone_takes_it_over_and_abandons_it(): + from app.engine.runner import run_manager + + async with SessionLocal() as session: + run = Run(status="running", owner=GHOST, graph={}, input={}) + session.add(run) + await session.commit() + assert await run_manager.cancel(run.id, actor="张工") == "cancelled" + async with SessionLocal() as session: + row = await session.get(Run, run.id) + assert row.status == "cancelled" and row.end_reason == "stopped" + + +async def test_leadership_is_handed_over_when_the_lease_runs_out(): + me = leader_mod.Leadership("test-lead") + assert await me.claim() and me.held() + first = me.epoch + async with SessionLocal() as session: + row = await session.get(ClusterLease, "test-lead") + row.owner, row.lease_until = PEER, utcnow() + timedelta(seconds=60) + await session.commit() + assert not await me.claim() and not me.held(), "别的进程拿着没到期的租约：不当班" + async with SessionLocal() as session: + row = await session.get(ClusterLease, "test-lead") + row.lease_until = utcnow() - timedelta(seconds=1) + await session.commit() + assert await me.claim() and me.epoch == first + 1, "到期了接班，换一次当班加一" + await me.release() + assert (await leader_mod.current("test-lead"))["owner"] is None + + +async def test_the_daily_cap_is_not_overrun_by_concurrent_charges(): + from app.engine.spend import Ledger + + book = Ledger(f"test_spend_{new_id()[:8]}") + other = Ledger(book.key) # 同一本账的另一个实例（另一个进程里就是这样） + got = await asyncio.gather(*(b.charge(0.1, limit=1.0, priced=True) for b in [book, other] * 10), + *(book.charge(0.1, limit=1.0, priced=True) for _ in range(5))) + assert sum(1 for d in got if d) == 10 + assert (await book.daily())["usd"] <= 1.0 + 1e-9 + + +@pytest.mark.sqlite_only +def test_sqlite_refuses_a_second_process_on_the_same_data_dir(): + from app.cluster import guard + + guard.acquire() + try: + backend = Path(__file__).resolve().parents[1] + env = {**os.environ, "PYTHONPATH": str(backend)} + out = subprocess.run([sys.executable, "-c", "from app.cluster import guard; guard.acquire()"], cwd=str(backend), + env=env, capture_output=True, text=True, timeout=60) + assert out.returncode != 0 and "只支持单进程" in out.stderr, out.stderr[-2000:] + many = subprocess.run([sys.executable, "-c", "from app.cluster import guard; guard.acquire()"], cwd=str(backend), + env={**env, "AGENTLAB_WORKERS": "2", "AGENTLAB_DATA_DIR": str(Path(env["AGENTLAB_DATA_DIR"]) / "other")}, + capture_output=True, text=True, timeout=60) + assert many.returncode != 0 and "AGENTLAB_WORKERS=2" in many.stderr, many.stderr[-2000:] + finally: + guard.release() + + +async def _until(check, timeout: float = 10.0) -> None: + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while not check(): + if loop.time() > deadline: + raise AssertionError("等超时了") + await asyncio.sleep(0.05)