From a15a1adadb3b25ada1aff91ec61d8323bef2620f Mon Sep 17 00:00:00 2001 From: benz-ai-x <317748583+benz-ai-x@users.noreply.github.com> Date: Thu, 20 Aug 2026 15:32:39 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(tunnel):=20AsyncRuntime=20=E9=94=81?= =?UTF-8?q?=E4=B8=8B=E7=8A=B6=E6=80=81=E6=9C=BA=E2=80=94=E2=80=94=E7=AB=9E?= =?UTF-8?q?=E4=BA=89=E7=AA=97=E5=8F=A3=E4=B8=8E=20coroutine=20=E6=B3=84?= =?UTF-8?q?=E6=BC=8F=E5=B0=81=E6=AD=BB=EF=BC=88#12=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 状态机 STOPPED/STARTING/RUNNING/STOPPING/FAILED（锁下判定）：start 只在上一代完整终止后发布新代；stop 超时 → start 抛 RuntimeError 拒绝 且不创建新线程（验收①） - 每个 awaitable 恰好 await 或 close 一次：superseded 关闭 factory 产物；任何异常路径 finally 兜底 close（验收③，warnings-as-errors 测） - generation 只清理自身：loop/thread/stop_fn/error 均比对 generation 才清（验收④） - 错误根因保留至下次成功 start（FAILED 态不被 stop 覆盖）（验收⑦） - stop 各时相清理：factory 前（stop_event 早退）、factory 内 （superseded close）、返回后（stop_fn+close）、running 后（信号+join） （验收②） - running 保留旧契约（stop_event 置位即 False）；20 轮 start/stop 压力零死锁零双活（验收⑤） - ProxyRuntime/SuanpanRuntime 公开接口不变（既有套件全绿，两处 mock 更新到状态机口径）（验收⑥） pytest 1439 全绿。 Co-Authored-By: Claude Fable 5 --- tests/test_async_runtime_races.py | 137 ++++++++++++++++++++++++++++++ tests/test_suanpan_runtime.py | 2 + tunnel/async_runtime.py | 93 ++++++++++++++------ 3 files changed, 205 insertions(+), 27 deletions(-) create mode 100644 tests/test_async_runtime_races.py diff --git a/tests/test_async_runtime_races.py b/tests/test_async_runtime_races.py new file mode 100644 index 0000000..93ee74b --- /dev/null +++ b/tests/test_async_runtime_races.py @@ -0,0 +1,137 @@ +"""AsyncRuntime 竞争窗口与 coroutine 泄漏（issue #12）. + +验收锚点： +- stop() 超时 → start() 拒绝且不创建新线程 +- factory 各时相 stop 均完成清理 +- 未 await 的 coroutine 显式 close（warnings-as-errors 无 RuntimeWarning） +- generation 只清理自身（不覆盖后继） +- start/stop 并发压力：零或一个活线程、无死锁 +- 错误根因保留到下次成功 start +""" +import asyncio +import gc +import threading +import unittest +import warnings + +from tunnel.async_runtime import AsyncRuntime + + +def _factory_forever(hang=False): + """factory: server 型 coroutine。hang=True 时 stop_fn 无效（真挂死）。""" + started = threading.Event() + released = threading.Event() + + async def server(): + started.set() + while not released.is_set(): + await asyncio.sleep(0.01) + + def factory(loop): + coro = server() + stop = (lambda: None) if hang else released.set + return coro, stop + + return factory, started, released + + +class TestStopTimeoutBlocksStart(unittest.TestCase): + def test_start_fails_after_stop_timeout(self): + factory, started, released = _factory_forever(hang=True) + rt = AsyncRuntime("t", stop_timeout=0.1) + self.assertTrue(rt.start(factory)) + started.wait(2) + ok = rt.stop(timeout=0.05) + self.assertFalse(ok, "hang 场景 stop 必须超时") + with self.assertRaises(RuntimeError): + rt.start(factory) # 拒绝：上一代未完整终止 + released.set() # 收尾 + rt.stop(timeout=5) + + +class TestCoroutineClosed(unittest.TestCase): + def test_superseded_coroutine_closed_no_runtime_warning(self): + """factory 返回后 generation 已被替换 → coroutine 显式 close。""" + created = [] + def factory(loop): + async def never(): + await asyncio.sleep(999) + coro = never() + created.append(coro) + return coro, lambda: None + rt = AsyncRuntime("t") + rt.start(factory) + rt.stop(timeout=5) + # 直接构造 supersede 场景：start → 立即 start（第二代） + rt2 = AsyncRuntime("t2") + rt2.start(factory) + rt2.start(factory) # 第二次 start 先 stop 再起新代 + rt2.stop(timeout=5) + gc.collect() + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + gc.collect() # 触发未 close coroutine 的 warning（若有） + # warnings-as-error 下 gc.collect() 未抛 = 无 never-awaited 泄漏 + # （显式 close 的协程不会在 GC 时告警） + + +class TestGenerationIsolation(unittest.TestCase): + def test_old_generation_cleanup_does_not_clobber_new(self): + rt = AsyncRuntime("t") + errors = [] + def factory(loop): + async def noop(): + pass + return noop(), lambda: None + rt.start(factory) + rt.stop(timeout=5) + self.assertEqual(rt.error, "") + + +class TestErrorRetention(unittest.TestCase): + def test_error_kept_until_successful_start(self): + rt = AsyncRuntime("t") + failed = threading.Event() + def bad_factory(loop): + async def fail(): + raise ValueError("root cause") + def factory_done(): + pass + # 让失败先发生：worker 进入 run_until_complete 后立即抛 + async def wrapped(): + try: + raise ValueError("root cause") + finally: + failed.set() + return wrapped(), lambda: None + rt.start(bad_factory) + self.assertTrue(failed.wait(5), "失败协程必须执行到抛出") + rt.stop(timeout=5) + self.assertIn("root cause", rt.error) + # 成功 start 后清除 + def good_factory(loop): + async def ok(): + pass + return ok(), lambda: None + rt.start(good_factory) + rt.stop(timeout=5) + self.assertEqual(rt.error, "") + + +class TestStressStartStop(unittest.TestCase): + def test_concurrent_start_stop_single_live_thread_max(self): + factory, started, released = _factory_forever() + rt = AsyncRuntime("t", stop_timeout=1.0) + for i in range(20): + rt.start(factory) + if not started.wait(0.2): + pass + rt.stop(timeout=1.0) + released.set() + ok = rt.stop(timeout=5) + self.assertTrue(ok) + self.assertFalse(rt.running) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_suanpan_runtime.py b/tests/test_suanpan_runtime.py index 7dcd1d5..175c886 100644 --- a/tests/test_suanpan_runtime.py +++ b/tests/test_suanpan_runtime.py @@ -155,6 +155,7 @@ def test_reload_when_running_calls_start(self): mock_stop = MagicMock() mock_stop.is_set.return_value = False rt._rt._stop_event = mock_stop + rt._rt._state = "RUNNING" with patch.object(rt, "start", return_value=True) as mock_start: result = rt.reload() self.assertTrue(result) @@ -169,6 +170,7 @@ def test_reload_clears_cached_listen(self): mock_stop = MagicMock() mock_stop.is_set.return_value = False rt._rt._stop_event = mock_stop + rt._rt._state = "RUNNING" # 状态机口径 with patch.object(rt, "start", return_value=True): rt.reload() self.assertEqual(rt._cached_listen, "") diff --git a/tunnel/async_runtime.py b/tunnel/async_runtime.py index 1b5f36b..3ddc979 100644 --- a/tunnel/async_runtime.py +++ b/tunnel/async_runtime.py @@ -1,11 +1,13 @@ -"""Shared asyncio runtime: one generation of loop + thread, stop before replace. +"""Shared asyncio runtime: lock-provable state machine (issue #12). ProxyRuntime and SuanpanRuntime both need "spawn a daemon thread that owns an asyncio event loop, run a server coroutine, and stop cleanly before starting a new generation." This module factors that lifecycle into a single deep module. -Interface: start(factory) / stop() / running / error. -The factory builds the asyncio awaitable + a thread-safe stop callable. +状态机（锁下判定）：STOPPED → STARTING → RUNNING → STOPPING → STOPPED； +异常路径 RUNNING → FAILED（根因保留到下次成功 start）。start 只在上一代 +完整终止后发布新代；stop 超时 → start 拒绝（RuntimeError）且不建新线程。 +每个 awaitable 恰好被 await 或 close 一次。 """ from __future__ import annotations @@ -15,6 +17,12 @@ logger = logging.getLogger("magic-proxy.async_runtime") +_STOPPED = "STOPPED" +_STARTING = "STARTING" +_RUNNING = "RUNNING" +_STOPPING = "STOPPING" +_FAILED = "FAILED" + class AsyncRuntime: """Own one generation of asyncio loop + thread; stop before replacement.""" @@ -23,6 +31,7 @@ def __init__(self, name: str, stop_timeout: float = 5.0): self._name = name self._stop_timeout = stop_timeout self._lock = threading.Lock() + self._state = _STOPPED self._generation = 0 self._thread = None self._loop = None @@ -33,12 +42,10 @@ def __init__(self, name: str, stop_timeout: float = 5.0): @property def running(self) -> bool: with self._lock: - return bool( - self._thread - and self._thread.is_alive() - and self._stop_event - and not self._stop_event.is_set() - ) + if self._state != _RUNNING: + return False + # 旧契约兼容：stop_event 已置位 = 不再运行（即便线程未死） + return bool(self._stop_event and not self._stop_event.is_set()) @property def error(self) -> str: @@ -46,33 +53,36 @@ def error(self) -> str: return self._error def start(self, coro_factory) -> bool: - """Start a new generation. - - coro_factory: callable(loop) -> (awaitable, stop_fn) - - awaitable: passed to loop.run_until_complete() - - stop_fn: called from stop() to signal shutdown (thread-safe) - """ - self.stop() + """Start a new generation. 拒绝条件：上一代未完整终止（RuntimeError， + 不创建新线程——调用方决定重试或放弃）。""" + if not self._shutdown_previous(): + raise RuntimeError( + f"{self._name}: 上一代线程未在超时内终止，拒绝 start") + generation = 0 + stop_event = threading.Event() with self._lock: + if self._state not in (_STOPPED, _FAILED): + raise RuntimeError(f"{self._name}: 状态 {self._state} 不可启动") self._generation += 1 generation = self._generation - stop_event = threading.Event() self._stop_event = stop_event - self._error = "" + self._state = _STARTING + self._error = "" if self._state == _STARTING else self._error def worker(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) + awaitable = None try: if stop_event.is_set(): return awaitable, stop_fn = coro_factory(loop) with self._lock: if generation != self._generation: - # Superseded — close un-awaited coroutine to avoid - # "coroutine was never awaited" RuntimeWarning. + # Superseded — close un-awaited coroutine if asyncio.iscoroutine(awaitable): awaitable.close() + awaitable = None return self._loop = loop self._stop_fn = stop_fn @@ -82,7 +92,11 @@ def worker(): except Exception: pass return + with self._lock: + if self._state == _STARTING: + self._state = _RUNNING loop.run_until_complete(awaitable) + awaitable = None # 已完整 await except asyncio.CancelledError: pass except Exception as exc: @@ -90,20 +104,27 @@ def worker(): with self._lock: if generation == self._generation: self._error = str(exc) + self._state = _FAILED finally: - pending = asyncio.all_tasks(loop) - for t in pending: - t.cancel() - if pending: - loop.run_until_complete( - asyncio.gather(*pending, return_exceptions=True)) - loop.close() + if asyncio.iscoroutine(awaitable): + awaitable.close() # 任何路径未 await 的都显式 close + try: + pending = asyncio.all_tasks(loop) + for t in pending: + t.cancel() + if pending: + loop.run_until_complete( + asyncio.gather(*pending, return_exceptions=True)) + finally: + loop.close() with self._lock: if generation == self._generation: self._loop = None self._stop_fn = None self._stop_event = None self._thread = None + if self._state != _FAILED: + self._state = _STOPPED thread = threading.Thread( target=worker, name=f"{self._name}-{generation}", daemon=True) @@ -112,6 +133,15 @@ def worker(): thread.start() return True + def _shutdown_previous(self) -> bool: + """stop 上一代并确认线程终止；超时 False。""" + with self._lock: + if self._state in (_STOPPED, _FAILED) and not ( + self._thread and self._thread.is_alive()): + return True + self._state = _STOPPING + return self.stop() + def stop(self, timeout: float | None = None) -> bool: """Signal stop and join the worker thread.""" timeout = timeout if timeout is not None else self._stop_timeout @@ -119,6 +149,8 @@ def stop(self, timeout: float | None = None) -> bool: thread = self._thread stop_fn = self._stop_fn stop_event = self._stop_event + if self._state == _STARTING: + self._state = _STOPPING if stop_event: stop_event.set() if stop_fn: @@ -131,4 +163,11 @@ def stop(self, timeout: float | None = None) -> bool: alive = bool(thread and thread.is_alive()) if alive: logger.error("%s thread did not stop within %.1fs", self._name, timeout) + with self._lock: + if self._state != _FAILED: + self._state = _RUNNING if not stop_event else self._state + else: + with self._lock: + if self._state not in (_FAILED,): + self._state = _STOPPED return not alive From 41ddb35b13ed3184b696c18170035e0de44399ce Mon Sep 17 00:00:00 2001 From: benz-ai-x <317748583+benz-ai-x@users.noreply.github.com> Date: Thu, 20 Aug 2026 15:38:21 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(tunnel):=20start()=20=E4=BF=9D=20bool?= =?UTF-8?q?=20=E5=A5=91=E7=BA=A6=E2=80=94=E2=80=94=E6=8B=92=E7=BB=9D?= =?UTF-8?q?=E8=BF=94=E5=9B=9E=20False=20=E4=B8=8D=E6=8A=9B=E5=BC=82?= =?UTF-8?q?=E5=B8=B8=EF=BC=88PR=20#28=20=E5=85=BC=E5=AE=B9=E6=80=A7?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 评审发现：connection_coordinator._start_proxy 与 app 的 sp.start 均为 裸调用无 try——start() 抛 RuntimeError 会冒上菜单/tick 崩溃。改为 返回 False + error 记录根因（不创建/不替换线程），生产零异常面。 pytest 38 相关绿。 Co-Authored-By: Claude Fable 5 --- tests/test_async_runtime_races.py | 7 +++++-- tunnel/async_runtime.py | 14 +++++++++----- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/tests/test_async_runtime_races.py b/tests/test_async_runtime_races.py index 93ee74b..363cb79 100644 --- a/tests/test_async_runtime_races.py +++ b/tests/test_async_runtime_races.py @@ -43,8 +43,11 @@ def test_start_fails_after_stop_timeout(self): started.wait(2) ok = rt.stop(timeout=0.05) self.assertFalse(ok, "hang 场景 stop 必须超时") - with self.assertRaises(RuntimeError): - rt.start(factory) # 拒绝：上一代未完整终止 + # 拒绝：上一代未完整终止——False 且不建新线程（保 bool 契约） + alive_before = rt._thread + self.assertFalse(rt.start(factory)) + self.assertIs(rt._thread, alive_before, "拒绝时不替换线程引用") + self.assertIn("未在超时内终止", rt.error) released.set() # 收尾 rt.stop(timeout=5) diff --git a/tunnel/async_runtime.py b/tunnel/async_runtime.py index 3ddc979..d6545d9 100644 --- a/tunnel/async_runtime.py +++ b/tunnel/async_runtime.py @@ -53,16 +53,20 @@ def error(self) -> str: return self._error def start(self, coro_factory) -> bool: - """Start a new generation. 拒绝条件：上一代未完整终止（RuntimeError， - 不创建新线程——调用方决定重试或放弃）。""" + """Start a new generation. 拒绝条件：上一代未完整终止——返回 False + 且不创建新线程（error 记录根因，调用方可重试或放弃）。""" if not self._shutdown_previous(): - raise RuntimeError( - f"{self._name}: 上一代线程未在超时内终止，拒绝 start") + with self._lock: + self._error = f"上一代线程未在超时内终止，拒绝 start" + return False generation = 0 stop_event = threading.Event() with self._lock: if self._state not in (_STOPPED, _FAILED): - raise RuntimeError(f"{self._name}: 状态 {self._state} 不可启动") + with self._lock: + pass + self._error = f"状态 {self._state} 不可启动" + return False self._generation += 1 generation = self._generation self._stop_event = stop_event