From 2ac61337b8444b93a17d4bcc9a2a5d7bc991ea8b Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 08:41:18 +0800 Subject: [PATCH] fix(async): offload synchronous plugin methods --- app/runtime/execution.py | 25 ++++++ app/runtime/extensions/module/dispatcher.py | 4 +- app/runtime/extensions/plugin_manager.py | 10 +-- .../architecture/dependency-baseline.json | 5 +- tests/test_plugin_manager_async_methods.py | 71 +++++++++++++++ tests/test_runtime_execution.py | 90 +++++++++++++++++++ 6 files changed, 196 insertions(+), 9 deletions(-) create mode 100644 tests/test_plugin_manager_async_methods.py create mode 100644 tests/test_runtime_execution.py diff --git a/app/runtime/execution.py b/app/runtime/execution.py index 4a0ab2f55..48839734f 100644 --- a/app/runtime/execution.py +++ b/app/runtime/execution.py @@ -21,6 +21,31 @@ async def run_in_threadpool( return await run_sync(context.run, func, *args) +async def run_in_threadpool_to_completion( + func: Callable[..., Any], + *args: Any, + **kwargs: Any, +) -> Any: + """在线程调用取得终态后传播取消，避免提前释放仍在使用的执行容量。""" + worker_task = asyncio.create_task(run_in_threadpool(func, *args, **kwargs)) + cancellation: asyncio.CancelledError | None = None + while not worker_task.done(): + try: + await asyncio.wait({worker_task}) + except asyncio.CancelledError as error: + cancellation = cancellation or error + continue + try: + result = worker_task.result() + except Exception as error: + if cancellation is not None: + raise cancellation from error + raise + if cancellation is not None: + raise cancellation + return result + + def retry(ExceptionToCheck: Any, tries: int = 3, delay: int = 3, backoff: int = 2, logger: Any = None): """ diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py index 8435fdc5b..385379a1f 100644 --- a/app/runtime/extensions/module/dispatcher.py +++ b/app/runtime/extensions/module/dispatcher.py @@ -7,7 +7,7 @@ from collections.abc import Callable, Mapping from typing import Any, Protocol, cast from app.foundation.reflection import ObjectUtils -from app.runtime.execution import run_in_threadpool +from app.runtime.execution import run_in_threadpool_to_completion from app.runtime.log import logger from app.runtime.observability import observe_duration, record_metric from app.runtime.extensions.module.contracts import ( @@ -49,7 +49,7 @@ class ModuleInvocationDispatcher: plugin_error_handler: ModuleErrorHandler, system_error_handler: ModuleErrorHandler, rate_limit_handler: ModuleErrorHandler, - async_function_runner: AsyncFunctionRunner = run_in_threadpool, + async_function_runner: AsyncFunctionRunner = run_in_threadpool_to_completion, ) -> None: """保存模块目录和策略回调，不主动发现或创建任何运行时资源。""" self._module_catalog = module_catalog diff --git a/app/runtime/extensions/plugin_manager.py b/app/runtime/extensions/plugin_manager.py index 746b3a00d..c505f18bb 100644 --- a/app/runtime/extensions/plugin_manager.py +++ b/app/runtime/extensions/plugin_manager.py @@ -1,4 +1,4 @@ -import asyncio +import inspect import posixpath import threading from contextlib import contextmanager @@ -13,6 +13,7 @@ from app.schemas.plugin import PluginInstance, PluginRuntimeStatus from app.foundation.crypto import RSAUtils from app.foundation.singleton import Singleton from app.foundation.version import compare_version +from app.runtime.execution import run_in_threadpool_to_completion from app.runtime.log import logger from app.runtime.observability import observe_compat_facade from app.runtime.settings import RuntimeSettingsCompat @@ -937,7 +938,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): async def async_run_plugin_method(self, pid: str, method: str, *args, **kwargs) -> Any: """ - 异步运行插件方法 + 异步运行插件方法，同步实现经受控线程入口执行 :param pid: 插件ID :param method: 方法名 :param args: 参数 @@ -949,10 +950,9 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): if not hasattr(plugin, method): return None method_func = getattr(plugin, method) - if asyncio.iscoroutinefunction(method_func): + if inspect.iscoroutinefunction(method_func): return await method_func(*args, **kwargs) - else: - return method_func(*args, **kwargs) + return await run_in_threadpool_to_completion(method_func, *args, **kwargs) def get_plugin_ids(self) -> List[str]: """ diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 9ea776895..56b617d1a 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6451, - "edge_sha256": "9075d8717e384580cd6a41bc36685438770db4a8d8f18c57e6c494f32937113a", + "edge_count": 6452, + "edge_sha256": "fb19022826b8955ed7639c669083f8d1238795bf997e23b28f588d885610aee0", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -5626,6 +5626,7 @@ "app.runtime.extensions.plugin_manager -> app.foundation.version", "app.runtime.extensions.plugin_manager -> app.runtime", "app.runtime.extensions.plugin_manager -> app.runtime.events", + "app.runtime.extensions.plugin_manager -> app.runtime.execution", "app.runtime.extensions.plugin_manager -> app.runtime.extensions", "app.runtime.extensions.plugin_manager -> app.runtime.extensions.plugin", "app.runtime.extensions.plugin_manager -> app.runtime.extensions.plugin.access", diff --git a/tests/test_plugin_manager_async_methods.py b/tests/test_plugin_manager_async_methods.py new file mode 100644 index 000000000..dcd8965f1 --- /dev/null +++ b/tests/test_plugin_manager_async_methods.py @@ -0,0 +1,71 @@ +"""插件管理器异步方法的执行边界回归。""" + +from types import SimpleNamespace +from typing import Iterator +from unittest.mock import AsyncMock + +import pytest + +from app.foundation.singleton import Singleton +from app.sdk.plugins import PluginManager + + +@pytest.fixture +def plugin_manager() -> Iterator[PluginManager]: + """构造隔离的插件管理器实例。""" + Singleton._instances.pop((PluginManager, (), frozenset()), None) + manager = PluginManager() + yield manager + Singleton._instances.pop((PluginManager, (), frozenset()), None) + + +@pytest.mark.asyncio +async def test_async_run_plugin_method_offloads_sync_plugin_method( + plugin_manager: PluginManager, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """异步入口不得在事件循环内直接执行同步插件方法。""" + calls: list[tuple[str, int]] = [] + + def sync_method(value: int) -> int: + calls.append(("sync", value)) + return value + 1 + + plugin_manager.running_plugins["DemoPlugin"] = SimpleNamespace( + sync_method=sync_method, + ) + worker = AsyncMock(return_value=2) + monkeypatch.setattr( + "app.runtime.extensions.plugin_manager.run_in_threadpool_to_completion", + worker, + ) + + assert await plugin_manager.async_run_plugin_method( + "DemoPlugin", "sync_method", 1 + ) == 2 + worker.assert_awaited_once_with(sync_method, 1) + assert calls == [] + + +@pytest.mark.asyncio +async def test_async_run_plugin_method_keeps_async_plugin_method_on_loop( + plugin_manager: PluginManager, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """原生协程插件方法继续直接等待，不额外占用同步 worker。""" + async def async_method(value: int) -> int: + return value + 1 + + plugin_manager.running_plugins["DemoPlugin"] = SimpleNamespace( + async_method=async_method, + ) + worker = AsyncMock() + monkeypatch.setattr( + "app.runtime.extensions.plugin_manager.run_in_threadpool_to_completion", + worker, + ) + + assert await plugin_manager.async_run_plugin_method( + "DemoPlugin", "async_method", 1 + ) == 2 + worker.assert_not_awaited() diff --git a/tests/test_runtime_execution.py b/tests/test_runtime_execution.py new file mode 100644 index 000000000..532ec5c7a --- /dev/null +++ b/tests/test_runtime_execution.py @@ -0,0 +1,90 @@ +"""运行时同步 worker 的取消与容量合同回归。""" + +import asyncio +import threading + +import pytest +from anyio.to_thread import current_default_thread_limiter + +from app.runtime.execution import run_in_threadpool_to_completion + + +@pytest.mark.asyncio +async def test_threadpool_capacity_is_held_until_cancelled_call_finishes() -> None: + """调用方取消后，执行令牌必须由真实同步调用持有到终态。""" + limiter = current_default_thread_limiter() + original_capacity = limiter.total_tokens + release = threading.Event() + first_started = threading.Event() + second_started = threading.Event() + + def blocking_call(started: threading.Event) -> None: + started.set() + release.wait() + + limiter.total_tokens = 1 + first = asyncio.create_task( + run_in_threadpool_to_completion(blocking_call, first_started) + ) + second = None + try: + while not first_started.is_set(): + await asyncio.sleep(0) + + first.cancel() + await asyncio.sleep(0) + first.cancel() + await asyncio.sleep(0) + + assert first.done() is False + assert limiter.borrowed_tokens == 1 + + second = asyncio.create_task( + run_in_threadpool_to_completion(blocking_call, second_started) + ) + await asyncio.sleep(0.01) + assert second_started.is_set() is False + + release.set() + with pytest.raises(asyncio.CancelledError): + await first + await second + finally: + release.set() + if not first.done(): + await asyncio.gather(first, return_exceptions=True) + if second is not None and not second.done(): + await asyncio.gather(second, return_exceptions=True) + limiter.total_tokens = original_capacity + + +@pytest.mark.asyncio +async def test_cancelled_threadpool_call_preserves_worker_failure_as_cause() -> None: + """调用方取消优先返回，线程终态异常仍保留为诊断原因。""" + loop = asyncio.get_running_loop() + previous_handler = loop.get_exception_handler() + loop_errors: list[dict] = [] + release = threading.Event() + started = threading.Event() + + def failing_call() -> None: + started.set() + release.wait() + raise ValueError("worker failed") + + task = asyncio.create_task(run_in_threadpool_to_completion(failing_call)) + while not started.is_set(): + await asyncio.sleep(0) + + task.cancel() + release.set() + + loop.set_exception_handler(lambda _loop, context: loop_errors.append(context)) + try: + with pytest.raises(asyncio.CancelledError) as error_info: + await task + await asyncio.sleep(0) + finally: + loop.set_exception_handler(previous_handler) + assert isinstance(error_info.value.__cause__, ValueError) + assert loop_errors == []