mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-04 23:17:20 +08:00
fix(async): offload synchronous plugin methods
This commit is contained in:
@@ -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):
|
||||
"""
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]:
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user