diff --git a/backend/app/cluster/slots.py b/backend/app/cluster/slots.py index a7c83136..d363dedb 100644 --- a/backend/app/cluster/slots.py +++ b/backend/app/cluster/slots.py @@ -194,6 +194,8 @@ class _Holder: run_id: str loop: asyncio.AbstractEventLoop woke: asyncio.Event = field(default_factory=asyncio.Event) + #: 还在排队等名额(发下来之前) + queued: bool = True _holders: dict[str, _Holder] = {} @@ -379,6 +381,7 @@ async def hold(run_id: str, origin_kind: str | None) -> AsyncIterator[str]: await grant() got = await _granted(run_id) if got: + holder.queued = False break if got is None: await _request(run_id, klass, priority) @@ -397,6 +400,17 @@ async def hold(run_id: str, origin_kind: str | None) -> AsyncIterator[str]: signals.publish("slot") +async def waiting(run_id: str, *, here: bool = False) -> bool: + """这次运行是不是正排着队等名额(评测每题的时限、算期望的时限不算这一段,eval/runner、eval/gold)。 + here:这次运行的执行任务在本进程里(还没开始要、或者已经放掉了名额的就不是在等,不用查库)。""" + holder = _holders.get(run_id) + if holder is not None: + return holder.queued + if here: + return False + return (await _granted(run_id)) is False + + async def snapshot() -> dict[str, Any]: """现在各类在跑、在等几个(系统信息、验收脚本看)。""" if _in_memory(): @@ -413,4 +427,5 @@ async def snapshot() -> dict[str, Any]: __all__ = ["CLASSES", "CLASS_LABEL", "ORIGIN_CLASS", "SETTING_KEY", "borrowable", "class_of", "defaults", "forget_policy", "grant", + "waiting", "hold", "merged", "policy", "problem", "snapshot", "values", "wake_local"] diff --git a/backend/app/core/job_runtime.py b/backend/app/core/job_runtime.py index a9f103e6..b050ee83 100644 --- a/backend/app/core/job_runtime.py +++ b/backend/app/core/job_runtime.py @@ -241,16 +241,40 @@ class SourceLanes: await session.execute(text("SELECT pg_advisory_xact_lock(:k)"), {"k": advisory_key(f"lane:{source_id}")}) - async def _rows(self, session: AsyncSession, source_id: str) -> list[Any]: - """这个源队里活着的各行(号小的在前);属主不在了的删掉(不提交)。""" - from app.cluster import state + async def _dead_owners(self, source_id: str) -> frozenset[str]: + """拿这个源的锁之前备好:队里别的进程的属主哪些确实不在了(手上那份不认识的当场再查一次登记)。持着咨询锁时不另取连接 + 查库(复核:连接池用尽时会卡到取连接超时);拿锁之后才出现的属主不在这里面,这一次当活着,下一次再说。""" + from app.cluster import identity, state + from app.db.models_cluster import LaneTicket as Row + + async with SessionLocal() as session: + owners = set((await session.execute(select(Row.owner).where(Row.source_id == source_id).distinct())) + .scalars()) + me = identity.owner() + others = {o for o in owners if o and o != me} + if not others: + return frozenset() + live = await state.live_for(others) + return frozenset(o for o in others if o not in live.others) + + async def _rows(self, session: AsyncSession, source_id: str, dead_owners: frozenset[str]) -> list[Any]: + """这个源队里活着的各行(号小的在前);属主不在了的删掉(不提交)。dead_owners 是拿锁之前备好的那一份:属主是别的进程的 + 只认它;属主是自己的看本进程里登记着没有(正有人排进来时一律算活着)。""" + from app.cluster import identity from app.db.models_cluster import LaneTicket as Row rows = list((await session.execute(select(Row).where(Row.source_id == source_id).order_by(Row.id))).scalars()) - live = await state.live_for([r.owner for r in rows]) - if self._entering.get(source_id): - live = live.trusting() - dead = [r.id for r in rows if not live.alive(r.owner, local=self._local_alive(r.id))] + me = identity.owner() + entering = bool(self._entering.get(source_id)) + + def alive(row: Any) -> bool: + if not row.owner: + return False + if row.owner == me: + return entering or self._local_alive(row.id) + return row.owner not in dead_owners + + dead = [r.id for r in rows if not alive(r)] if dead: from sqlalchemy import delete @@ -262,9 +286,10 @@ class SourceLanes: from app.db.models_cluster import LaneTicket as Row granted_now: int | None = None + dead_owners = await self._dead_owners(source_id) async with self._lock(), SessionLocal() as session: await self._locked(session, source_id) - rows = await self._rows(session, source_id) + rows = await self._rows(session, source_id, dead_owners) head = next((r for r in rows if r.granted), None) if head is None and rows: head = rows[0] @@ -318,9 +343,10 @@ class SourceLanes: if kind not in _LANE_KINDS: raise ValueError(f"同源队里没有登记「{kind}」这种事,先 register_lane_kind") self._listen() + dead_owners = await self._dead_owners(source_id) async with self._lock(), SessionLocal() as session: await self._locked(session, source_id) - rows = await self._rows(session, source_id) + rows = await self._rows(session, source_id, dead_owners) if rows: head = next((r for r in rows if r.granted), rows[0]) await session.commit() @@ -337,8 +363,9 @@ class SourceLanes: async def holders(self, source_id: str) -> list[LaneHolder]: """队里的各位,排在最前面(正在做)的在前。""" + dead_owners = await self._dead_owners(source_id) async with SessionLocal() as session: - rows = await self._rows(session, source_id) + rows = await self._rows(session, source_id, dead_owners) await session.commit() rows.sort(key=lambda r: (not r.granted, r.id)) return [LaneHolder(r.kind, r.job_id, r.owner) for r in rows] diff --git a/backend/app/eval/answer.py b/backend/app/eval/answer.py index 6acc25cf..2842b27e 100644 --- a/backend/app/eval/answer.py +++ b/backend/app/eval/answer.py @@ -24,7 +24,7 @@ app/chat/job.py),生成过程是 api/copilot.answer_ops,几个「决定」 用量按角色记(做法 3、11):助手 = 生成(answer_ops 末尾的 usage)+ 复核(review_of_run 的 usage);写作 = 答案运行里的 llm.end(按模型分)。重来的话两次都算——那都是这一题花的钱。 -时限、停止由调用方管(runner.py 用 asyncio.wait_for 包住 drive):这里随时把正在跑的那次运行记在 current_run,调用方 +时限、停止由调用方管(runner.py 用 within 包住 drive,排队等名额的那一段不算):这里随时把正在跑的那次运行记在 current_run,调用方 超时、停止时据此取消它(stop_run)。 """ from __future__ import annotations diff --git a/backend/app/eval/gold.py b/backend/app/eval/gold.py index 872aaf32..4b584f1b 100644 --- a/backend/app/eval/gold.py +++ b/backend/app/eval/gold.py @@ -209,11 +209,19 @@ async def _launch_and_wait(graph: dict[str, Any], payload: dict[str, Any], *, na run_id = run.id except LaunchRefused as e: return _Ran("failed", None, f"算期望的运行没有发起:{e.message}", GOLD_FAILED) + from app.cluster import slots + loop = asyncio.get_running_loop() deadline = loop.time() + max(0.0, float(ctx.timeout_s)) + last = loop.time() while True: # 执行任务收完尾(含封存)才回来;任务已经不在了(刚建好还没排上、或者已经结束)就立刻回来,再按库里的状态判 - await run_manager.wait_idle(run_id, timeout=max(0.05, min(5.0, deadline - loop.time()))) + await run_manager.wait_idle(run_id, timeout=max(0.05, min(1.0, deadline - loop.time()))) + now = loop.time() + if await slots.waiting(run_id, here=run_manager.is_active(run_id)): + # 排队等名额的那一段不算时限(复核:满载时排队不该算进每题时限) + deadline += now - last + last = now async with SessionLocal() as session: row = await session.get(Run, run_id) status, error = (row.status, row.error) if row is not None else ("failed", "运行记录不见了") diff --git a/backend/app/eval/runner.py b/backend/app/eval/runner.py index 441430d7..af8718f7 100644 --- a/backend/app/eval/runner.py +++ b/backend/app/eval/runner.py @@ -15,7 +15,7 @@ snapshot 视图):起草、剖析照常进行,也不影响这次评测的 **并发。** 同时最多 max_parallel 题,且所有臂合起来不超过全局运行并发的四分之一。评测发起的作答、算期望的运行归「系统」类 (app/cluster/slots.py):全站的名额按类别有上限和优先级,评测再多也只占得到系统那一类的名额,问数据、试跑优先,不被挤占。 A/B 的同一题几臂同时开跑。每题时限 per_case_timeout_s:作答一臂超时就取消它正在跑的运行、记超时; -期望探针每次也按这个时限。 +期望探针每次也按这个时限。排队等运行名额的那一段不算时限(within、gold._launch_and_wait):满载时排队不该被记成超时、触发止损。 **预算**(评测-D9;单独一本账 engine/spend.eval_spend,不挤占裁判、核查的额度):开跑前按上一次的每题平均花费估算(estimate)。 每题开评前按估算预留——加上会超过这次评测的金额上限、或评测每日上限就不再开新题,剩下的记「已到上限」;评完按实际找补。 @@ -33,6 +33,7 @@ A/B 的同一题几臂同时开跑。每题时限 per_case_timeout_s:作答一 from __future__ import annotations import asyncio +import contextlib import copy import logging from collections.abc import Mapping, Sequence @@ -519,9 +520,9 @@ class EvalExecution(runtime.JobRun): try: if case.get("kind") == "workflow": graph, inputs = await self._workflow_graph(case, values) - await asyncio.wait_for(driver.drive_workflow(graph, inputs), timeout=timeout) + await within(driver.drive_workflow(graph, inputs), timeout, lambda: _queued(driver)) else: - await asyncio.wait_for(driver.drive(), timeout=timeout) + await within(driver.drive(), timeout, lambda: _queued(driver)) except asyncio.TimeoutError: turn = driver.turn turn.timed_out = True @@ -964,6 +965,40 @@ def running(run: Any) -> bool: return manager.running(leader_of(run)) +async def _queued(driver: Any) -> bool: + """这一题的作答运行正排着队等名额(app/cluster/slots):这一段不算每题的时限。""" + from app.cluster import slots + from app.engine.runner import run_manager + + run_id = getattr(driver.turn, "current_run", None) + return bool(run_id) and await slots.waiting(run_id, here=run_manager.is_active(run_id)) + + +async def within(coro: Any, budget: float, paused: Any, *, tick: float = 0.5) -> Any: + """同 asyncio.wait_for(coro, budget),只是 paused() 为真的那几段不算时间(复核:满载时排队等名额不该算进每题时限、触发止损)。 + 超时取消 coro、抛 TimeoutError。""" + loop = asyncio.get_running_loop() + task = asyncio.ensure_future(coro) + spent, last = 0.0, loop.time() + try: + while True: + done, _ = await asyncio.wait({task}, timeout=max(0.02, min(tick, budget - spent))) + if task in done: + return task.result() + now = loop.time() + step, last = now - last, now + if not await paused(): + spent += step + if spent >= budget: + task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + raise asyncio.TimeoutError() + except asyncio.CancelledError: + task.cancel() + raise + + def leader_of_id(rows: Sequence[EvalRun], job_id: str) -> str: """rows 里 id 是 job_id 的那一臂由哪一臂的任务在跑(A/B 几臂共用领头那一臂的后台任务)。""" row = next((r for r in rows if r.id == job_id), None) diff --git a/backend/tests/test_cluster.py b/backend/tests/test_cluster.py index 2243dc6f..58c6564a 100644 --- a/backend/tests/test_cluster.py +++ b/backend/tests/test_cluster.py @@ -230,6 +230,54 @@ async def test_the_source_lane_is_shared_and_skips_tickets_of_a_gone_process(): await other.release() +async def test_the_source_lane_does_not_look_up_liveness_while_holding_its_lock(monkeypatch): + """复核:持着这个源的咨询锁时不另取连接查「谁还活着」(连接池用尽时会卡到取连接超时)——拿锁之前备好;拿锁之后才出现的 + 别的属主这一次当活着,不删它的行。""" + from app.core import job_runtime as runtime + + import app.data.catalog_jobs # noqa: F401 - 登记 draft、profile + + lanes = runtime.lanes + holding = {"now": False, "looked_inside": 0} + real_locked, real_rows, real_live_for = lanes._locked, lanes._rows, state.live_for + + async def locked(session, source_id): + await real_locked(session, source_id) + holding["now"] = True + + async def rows(session, source_id, dead_owners): + try: + return await real_rows(session, source_id, dead_owners) + finally: + holding["now"] = False + + async def live_for(owners): + if holding["now"]: + holding["looked_inside"] += 1 + return await real_live_for(owners) + + monkeypatch.setattr(lanes, "_locked", locked) + monkeypatch.setattr(lanes, "_rows", rows) + monkeypatch.setattr(state, "live_for", live_for) + async with SessionLocal() as session: + session.add(LaneTicket(source_id="src-lock", kind="profile", job_id="ghost-job", owner=GHOST, granted=True)) + await session.commit() + mine = await lanes.enter("src-lock", "draft", "job-lock") + assert mine.granted.is_set(), "没了的进程占着的位置照旧让出来(拿锁之前就查清楚了)" + assert holding["looked_inside"] == 0 + await mine.release() + + async with SessionLocal() as session: + session.add(LaneTicket(source_id="src-late", kind="profile", job_id="late-job", owner=GHOST, granted=True)) + await session.commit() + kept = await real_rows(session, "src-late", frozenset()) + await session.commit() + assert [r.owner for r in kept] == [GHOST], "拿锁之前那份里没有的属主这一次当活着" + async with SessionLocal() as session: + await session.execute(delete(LaneTicket).where(LaneTicket.source_id.in_(["src-lock", "src-late"]))) + await session.commit() + + async def test_reaping_runs_only_touches_the_ones_whose_owner_is_gone(): from app.engine.runner import run_manager diff --git a/backend/tests/test_eval_runner.py b/backend/tests/test_eval_runner.py index a9685e62..94a9c106 100644 --- a/backend/tests/test_eval_runner.py +++ b/backend/tests/test_eval_runner.py @@ -289,6 +289,87 @@ async def test_a_case_over_its_time_limit_stops_its_run(monkeypatch): assert item.status == "ok" and item.path == "build_failed" and "每题时限" in item.error +async def test_time_queued_for_a_run_slot_does_not_count_against_the_case_limit(monkeypatch): + """复核:满载时作答运行排队等名额的那一段不算每题时限(排了 0.8 秒、时限 0.3 秒,拿到名额后很快答完:不记超时)。""" + from app.cluster import slots + + stored = slots.values({"max_total": 2, "interactive_reserve": 1}) + + async def fixed(*, fresh: bool = False): # noqa: ARG001 + return stored + + monkeypatch.setattr(slots, "policy", fixed) + slots.forget_policy() + gate = asyncio.Event() + + async def blocker() -> None: + async with slots.hold("eval-slot-blocker", "eval"): + await gate.wait() + + async def drive(self): + self.turn.current_run = "eval-answer-in-queue" + await self._phase("build") + async with slots.hold(self.turn.current_run, "eval"): + await asyncio.sleep(0.05) + self.turn.path = "asked" + self.turn.usage["assistant"] = {"by_model": {}, "input_tokens": 0, "output_tokens": 0, "calls": 1, + "cost_usd": 0.0, "unpriced_calls": 0} + return self.turn + + monkeypatch.setattr(answer_mod.Driver, "drive", drive) + original = runner.EvalExecution._prepare + + async def prepare(self): + units = await original(self) + self.budget["per_case_timeout_s"] = 0.3 + return units + + monkeypatch.setattr(runner.EvalExecution, "_prepare", prepare) + held = asyncio.create_task(blocker()) + try: + while not (await slots.snapshot())["classes"]["system"]["running"]: + await asyncio.sleep(0.02) + suite_id = await es.make_suite("none", cases=ask_cases(1)) + [run] = await start(suite_id) + while not await slots.waiting("eval-answer-in-queue"): + await asyncio.sleep(0.02) + await asyncio.sleep(0.8) + gate.set() + await es.wait_eval(run.id) + finally: + gate.set() + await held + [item] = await results(run.id) + assert item.status == "ok" and item.path == "asked" and not item.error, item.error + + +async def test_within_does_not_count_the_paused_stretches(): + paused = {"now": True} + + async def is_paused() -> bool: + return paused["now"] + + async def work() -> str: + await asyncio.sleep(0.5) + paused["now"] = False + await asyncio.sleep(0.05) + return "done" + + assert await runner.within(work(), 0.2, is_paused, tick=0.02) == "done", "暂停的那 0.5 秒不算" + + async def never(): + await asyncio.sleep(10) + + async def awake() -> bool: + return False + + loop = asyncio.get_running_loop() + t0 = loop.time() + with pytest.raises(asyncio.TimeoutError): + await runner.within(never(), 0.2, awake, tick=0.02) + assert loop.time() - t0 < 1.0 + + async def test_a_run_stopped_at_an_approval_is_cancelled_not_approved(client, monkeypatch, tmp_path): from app.eval.answer import APPROVAL_REASON