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()