From b52a67b18d31dcebeceb0ef340adcfce515ad06f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 15:29:00 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8E=E5=8F=B0=E4=BB=BB=E5=8A=A1=E7=9A=84?= =?UTF-8?q?=20drain=20=E4=B8=8D=E5=86=8D=E7=A9=BA=E8=BD=AC=EF=BC=9Anotify.?= =?UTF-8?q?service=E3=80=81eval.cleanup=20=E7=9A=84=20drain=20=E6=94=B9?= =?UTF-8?q?=E7=94=A8=E6=96=B0=E5=A2=9E=E7=9A=84=20core/background.drain?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=81=9A=E5=AE=8C=E4=BA=86=E7=9A=84=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E7=9B=B4=E6=8E=A5=E4=BB=8E=E9=9B=86=E5=90=88=E9=87=8C?= =?UTF-8?q?=E6=91=98=E6=8E=89=EF=BC=8C=E5=8F=AA=E7=AD=89=E6=9C=AC=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=BE=AA=E7=8E=AF=E4=B8=8A=E6=B2=A1=E5=81=9A=E5=AE=8C?= =?UTF-8?q?=E7=9A=84=EF=BC=8C=E5=88=AB=E7=9A=84=E5=BE=AA=E7=8E=AF=E4=B8=8A?= =?UTF-8?q?=E6=B2=A1=E5=81=9A=E5=AE=8C=E7=9A=84=E4=B8=8D=E7=AD=89=E3=80=81?= =?UTF-8?q?=E5=BE=AA=E7=8E=AF=E5=85=B3=E4=BA=86=E7=9A=84=E6=91=98=E6=8E=89?= =?UTF-8?q?=E3=80=82=E4=BB=A5=E5=89=8D=20while=20=5Fbackground:=20await=20?= =?UTF-8?q?gather(...)=20=E7=A2=B0=E4=B8=8A=E5=88=AB=E7=9A=84=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=BE=AA=E7=8E=AF=E4=B8=8A=E5=81=9A=E5=AE=8C=E3=80=81?= =?UTF-8?q?discard=20=E5=9B=9E=E8=B0=83=E6=B0=B8=E8=BF=9C=E4=B8=8D?= =?UTF-8?q?=E8=B7=91=E7=9A=84=E4=BB=BB=E5=8A=A1=EF=BC=8Cgather=20=E7=AB=8B?= =?UTF-8?q?=E5=88=BB=E8=BF=94=E5=9B=9E=E3=80=81=E4=B8=8D=E8=AE=A9=E5=87=BA?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E5=BE=AA=E7=8E=AF=EF=BC=8C=E5=9C=A8=E9=82=A3?= =?UTF-8?q?=E9=87=8C=E7=A9=BA=E8=BD=AC=EF=BC=88CPU=20=E6=89=93=E6=BB=A1?= =?UTF-8?q?=E3=80=81=E8=BF=9B=E7=A8=8B=E5=8D=A1=E6=AD=BB=EF=BC=89=EF=BC=9A?= =?UTF-8?q?PostgreSQL=20=E5=85=A8=E9=87=8F=E9=87=8C=20test=5Fpersonal=5Fap?= =?UTF-8?q?ps=20=E7=9A=84=E6=94=B6=E5=B0=BE=E4=B8=A4=E6=AC=A1=E5=8D=A1?= =?UTF-8?q?=E5=9C=A8=E8=BF=99=E9=87=8C=E3=80=82=E5=8D=95=E6=B5=8B=E5=9C=A8?= =?UTF-8?q?=E5=8F=A6=E4=B8=80=E4=B8=AA=E7=BA=BF=E7=A8=8B=E9=87=8C=E8=B7=91?= =?UTF-8?q?=20drain=E3=80=81=E6=8C=89=E7=BA=BF=E7=A8=8B=E7=AD=89=EF=BC=88?= =?UTF-8?q?=E7=A9=BA=E8=BD=AC=E6=97=B6=E4=BA=8B=E4=BB=B6=E5=BE=AA=E7=8E=AF?= =?UTF-8?q?=E8=BD=AC=E4=B8=8D=E5=8A=A8=EF=BC=8Cwait=5Ffor=20=E6=95=91?= =?UTF-8?q?=E4=B8=8D=E4=BA=86=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/core/background.py | 28 +++++++++ backend/app/eval/cleanup.py | 6 +- backend/app/notify/service.py | 6 +- backend/tests/test_background_drain.py | 81 ++++++++++++++++++++++++++ 4 files changed, 117 insertions(+), 4 deletions(-) create mode 100644 backend/app/core/background.py create mode 100644 backend/tests/test_background_drain.py diff --git a/backend/app/core/background.py b/backend/app/core/background.py new file mode 100644 index 00000000..2f47ea7c --- /dev/null +++ b/backend/app/core/background.py @@ -0,0 +1,28 @@ +"""进程里「发出去就不管」的后台任务（推送通知、评测收尾的清理）：留着引用免得被回收，测试和关停时等它们做完。""" + +from __future__ import annotations + +import asyncio +from typing import Any + + +async def drain(tasks: set[asyncio.Future[Any]]) -> None: + """等 tasks 里本事件循环上的任务都做完。 + + 已经做完的直接从集合里摘掉，不等它们自己的 discard 回调：done 回调排在任务所在的事件循环上，那个循环要是已经不转了（测试里 + 每个测试一个事件循环，前一个测试末尾刚做完的推送）回调就永远不跑、任务一直留在集合里。以前的写法 `while tasks: await + gather(*tasks)` 碰上这种集合，gather 拿到的全是做完了的、立刻返回，await 不让出事件循环，就在这里空转——CPU 打满、整个 + 进程卡死（PostgreSQL 全量里 test_personal_apps 收尾时出现过）。别的事件循环上还没做完的等不到（不在这个循环上），不等； + 那个循环已经关了的一并摘掉。""" + loop = asyncio.get_running_loop() + while True: + for task in list(tasks): + if task.done() or (task.get_loop() is not loop and task.get_loop().is_closed()): + tasks.discard(task) + mine = [task for task in tasks if task.get_loop() is loop] + if not mine: + return + await asyncio.gather(*mine, return_exceptions=True) + + +__all__ = ["drain"] diff --git a/backend/app/eval/cleanup.py b/backend/app/eval/cleanup.py index a52a947e..956b37f2 100644 --- a/backend/app/eval/cleanup.py +++ b/backend/app/eval/cleanup.py @@ -119,8 +119,10 @@ def spawn(suite_id: str) -> None: async def drain() -> None: """等后台的清理做完（测试、关停用）。""" - while _background: - await asyncio.gather(*list(_background), return_exceptions=True) + from app.core.background import drain as drain_tasks + + # 不写成 while _background: await gather(...)：别的事件循环上做完、还没摘掉的任务会让那样的循环空转（core/background） + await drain_tasks(_background) __all__ = ["clean_suite", "drain", "keep_set", "spawn"] diff --git a/backend/app/notify/service.py b/backend/app/notify/service.py index 17366503..211c35ff 100644 --- a/backend/app/notify/service.py +++ b/backend/app/notify/service.py @@ -121,8 +121,10 @@ def spawn(coro: Coroutine[Any, Any, Any]) -> None: async def drain() -> None: """等后台的推送都推完（测试、关停用）。""" - while _background: - await asyncio.gather(*list(_background), return_exceptions=True) + from app.core.background import drain as drain_tasks + + # 不写成 while _background: await gather(...)：别的事件循环上做完、还没摘掉的任务会让那样的循环空转（core/background） + await drain_tasks(_background) async def personal_url_problem(url: str) -> str | None: diff --git a/backend/tests/test_background_drain.py b/backend/tests/test_background_drain.py new file mode 100644 index 00000000..4aab03af --- /dev/null +++ b/backend/tests/test_background_drain.py @@ -0,0 +1,81 @@ +"""core/background.drain：等后台任务做完，碰上别的事件循环上做完、还没摘掉的任务不空转。 + +以前 notify.service.drain、eval.cleanup.drain 写的是 `while _background: await gather(*_background)`：前一个测试的事件循环上 +刚做完的推送，done 回调（从集合里摘掉）排在那个已经不转的循环上、永远不跑；gather 拿到的全是做完了的、立刻返回，await 不让出 +事件循环，于是空转、整个测试进程卡死（PostgreSQL 全量里 test_personal_apps 收尾时出现过两次）。空转时事件循环也转不动， +asyncio.wait_for 的超时救不了，所以这里在另一个线程里跑、按线程等。 +""" + +from __future__ import annotations + +import asyncio +import threading +from typing import Any + +import pytest + +from app.core import background +from app.eval import cleanup +from app.notify import service + + +def _finished_on_a_dead_loop(tasks: set[asyncio.Future[Any]]) -> asyncio.Future[Any]: + """在另一个事件循环上做完、但 discard 回调没跑过的任务：照 spawn 的写法登记，结果设上（回调排进那个循环），循环不转就关。""" + other = asyncio.new_event_loop() + fut = other.create_future() + tasks.add(fut) + fut.add_done_callback(tasks.discard) + fut.set_result(None) + other.close() + return fut + + +def _returns_within(seconds: float, make: Any) -> bool: + done = threading.Event() + + def go() -> None: + asyncio.run(make()) + done.set() + + threading.Thread(target=go, daemon=True).start() + return done.wait(seconds) + + +@pytest.mark.parametrize("module", [service, cleanup], ids=["notify", "eval-cleanup"]) +def test_drain_does_not_spin_on_tasks_finished_on_another_loop(module): + stale = _finished_on_a_dead_loop(module._background) + try: + assert _returns_within(5, module.drain), "drain 卡住了（空转）" + assert stale not in module._background, "做完了的摘掉" + finally: + module._background.discard(stale) + + +def test_drain_waits_for_this_loops_tasks_and_skips_other_loops_pending_ones(): + tasks: set[asyncio.Future[Any]] = set() + other = asyncio.new_event_loop() + pending_elsewhere = other.create_future() # 别的循环上没做完的：等不到，不等 + tasks.add(pending_elsewhere) + seen: list[str] = [] + + async def scenario() -> None: + async def slow() -> None: + await asyncio.sleep(0.05) + seen.append("slow") + + task = asyncio.get_running_loop().create_task(slow()) + tasks.add(task) + task.add_done_callback(tasks.discard) + await background.drain(tasks) + seen.append("drained") + + try: + assert _returns_within(5, scenario) + assert seen == ["slow", "drained"], "本循环上的等它做完" + assert pending_elsewhere in tasks, "循环还开着的别人的任务留着" + other.close() + assert _returns_within(5, lambda: background.drain(tasks)) + assert tasks == set(), "循环关了的一并摘掉" + finally: + if not other.is_closed(): + other.close()