复核「另外两处」:评测每题的时限、算期望的时限都不算排队等运行名额的那一段(within 代替 asyncio.wait_for,作答那次运行在 slots 里还排着队时不走表;算期望的运行排着队时把截止时刻往后挪),满载时不再被记成超时、触发止损;同源队(SourceLanes)跟发名额一样,拿这个源的咨询锁之前备好「哪些别的属主确实不在了」,持锁期间只用持锁的那条连接,拿锁之后才出现的属主这一次当活着。slots 加 waiting(run_id, here=):本进程里的直接看、不查库。补测试:排队 0.8 秒、时限 0.3 秒照样答完不记超时(改前的写法这条不过);within 暂停的那几段不算、不暂停照样超时;同源队持锁期间不查「谁还活着」、没了的进程照样让位、拿锁之后才出现的属主不删

This commit is contained in:
ThearchyHelios
2026-10-19 13:31:46 +08:00
parent f3ff5b9c18
commit 62089f2849
7 changed files with 229 additions and 15 deletions
+15
View File
@@ -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"]
+37 -10
View File
@@ -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]
+1 -1
View File
@@ -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
+9 -1
View File
@@ -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", "运行记录不见了")
+38 -3
View File
@@ -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)
+48
View File
@@ -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
+81
View File
@@ -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