From 54124979b0c8c87a60a2cb90031ff5abb56c581f Mon Sep 17 00:00:00 2001 From: Soulter <905617992@qq.com> Date: Mon, 6 Jul 2026 23:26:57 +0800 Subject: [PATCH] feat: add event loop diagnostics --- astrbot/core/core_lifecycle.py | 10 +- astrbot/core/utils/event_loop_diagnostics.py | 221 +++++++++++++++++++ tests/unit/test_event_loop_diagnostics.py | 109 +++++++++ 3 files changed, 339 insertions(+), 1 deletion(-) create mode 100644 astrbot/core/utils/event_loop_diagnostics.py create mode 100644 tests/unit/test_event_loop_diagnostics.py diff --git a/astrbot/core/core_lifecycle.py b/astrbot/core/core_lifecycle.py index c325a2ea3..215b81754 100644 --- a/astrbot/core/core_lifecycle.py +++ b/astrbot/core/core_lifecycle.py @@ -35,6 +35,9 @@ from astrbot.core.star.star_manager import PluginManager from astrbot.core.subagent_orchestrator import SubAgentOrchestrator from astrbot.core.umop_config_router import UmopConfigRouter from astrbot.core.updator import AstrBotUpdator +from astrbot.core.utils.event_loop_diagnostics import ( + create_event_loop_diagnostic_tasks, +) from astrbot.core.utils.llm_metadata import update_llm_metadata from astrbot.core.utils.migra_helper import migra from astrbot.core.utils.temp_dir_cleaner import TempDirCleaner @@ -296,13 +299,18 @@ class AstrBotCoreLifecycle: self.temp_dir_cleaner.run(), name="temp_dir_cleaner", ) + diagnostic_tasks = create_event_loop_diagnostic_tasks() # 把插件中注册的所有协程函数注册到事件总线中并执行 extra_tasks = [] for task in self.star_context._register_tasks: extra_tasks.append(asyncio.create_task(task, name=task.__name__)) # type: ignore - tasks_ = [event_bus_task, *(extra_tasks if extra_tasks else [])] + tasks_ = [ + event_bus_task, + *diagnostic_tasks, + *(extra_tasks if extra_tasks else []), + ] if cron_task: tasks_.append(cron_task) if temp_dir_cleaner_task: diff --git a/astrbot/core/utils/event_loop_diagnostics.py b/astrbot/core/utils/event_loop_diagnostics.py new file mode 100644 index 000000000..487fee061 --- /dev/null +++ b/astrbot/core/utils/event_loop_diagnostics.py @@ -0,0 +1,221 @@ +import asyncio +import faulthandler +import os +import sys +from dataclasses import dataclass +from typing import TextIO + +from astrbot import logger + +LAG_MONITOR_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_LAG_MONITOR" +LAG_MONITOR_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_LAG_INTERVAL" +LAG_MONITOR_THRESHOLD_ENV = "ASTRBOT_EVENT_LOOP_LAG_THRESHOLD" +WATCHDOG_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG" +WATCHDOG_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_INTERVAL" +WATCHDOG_TIMEOUT_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_TIMEOUT" + +DEFAULT_LAG_MONITOR_INTERVAL = 1.0 +DEFAULT_LAG_MONITOR_THRESHOLD = 5.0 +DEFAULT_WATCHDOG_INTERVAL = 1.0 +DEFAULT_WATCHDOG_TIMEOUT = 15.0 + + +@dataclass(frozen=True) +class EventLoopDiagnosticSettings: + """Settings for event loop lag and blockage diagnostics. + + Args: + lag_monitor_enabled: Whether to log event loop scheduling lag. + lag_monitor_interval: Seconds between lag monitor wakeups. + lag_monitor_threshold: Minimum lag seconds before logging a warning. + watchdog_enabled: Whether to arm the faulthandler watchdog. + watchdog_interval: Seconds between faulthandler watchdog refreshes. + watchdog_timeout: Seconds without event loop progress before dumping stacks. + """ + + lag_monitor_enabled: bool + lag_monitor_interval: float + lag_monitor_threshold: float + watchdog_enabled: bool + watchdog_interval: float + watchdog_timeout: float + + +def _env_flag(name: str, default: bool) -> bool: + """Read a boolean flag from the environment. + + Args: + name: Environment variable name. + default: Value to use when the variable is unset or empty. + + Returns: + Parsed boolean value. + """ + value = os.environ.get(name) + if value is None or value.strip() == "": + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + +def _env_float(name: str, default: float, minimum: float) -> float: + """Read a bounded float from the environment. + + Args: + name: Environment variable name. + default: Value to use when parsing fails or the value is too small. + minimum: Smallest accepted value. + + Returns: + Parsed float value or the default. + """ + value = os.environ.get(name) + if value is None or value.strip() == "": + return default + try: + parsed = float(value) + except ValueError: + logger.warning( + "Invalid %s=%r, fallback to %.3fs.", + name, + value, + default, + ) + return default + if parsed < minimum: + logger.warning( + "Invalid %s=%r, expected at least %.3fs; fallback to %.3fs.", + name, + value, + minimum, + default, + ) + return default + return parsed + + +def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings: + """Load event loop diagnostic settings from environment variables. + + Returns: + Event loop diagnostic settings. + """ + return EventLoopDiagnosticSettings( + lag_monitor_enabled=_env_flag(LAG_MONITOR_ENABLED_ENV, True), + lag_monitor_interval=_env_float( + LAG_MONITOR_INTERVAL_ENV, + DEFAULT_LAG_MONITOR_INTERVAL, + 0.1, + ), + lag_monitor_threshold=_env_float( + LAG_MONITOR_THRESHOLD_ENV, + DEFAULT_LAG_MONITOR_THRESHOLD, + 0.1, + ), + watchdog_enabled=_env_flag(WATCHDOG_ENABLED_ENV, False), + watchdog_interval=_env_float( + WATCHDOG_INTERVAL_ENV, + DEFAULT_WATCHDOG_INTERVAL, + 0.1, + ), + watchdog_timeout=_env_float( + WATCHDOG_TIMEOUT_ENV, + DEFAULT_WATCHDOG_TIMEOUT, + 1.0, + ), + ) + + +async def monitor_event_loop_lag( + *, + interval: float = DEFAULT_LAG_MONITOR_INTERVAL, + warn_after: float = DEFAULT_LAG_MONITOR_THRESHOLD, +) -> None: + """Log a warning when the event loop wakes significantly later than expected. + + Args: + interval: Seconds between monitor wakeups. + warn_after: Minimum lag seconds before logging a warning. + """ + loop = asyncio.get_running_loop() + expected = loop.time() + interval + while True: + await asyncio.sleep(interval) + now = loop.time() + lag = now - expected + if lag > warn_after: + logger.warning( + "Event loop lag detected: %.3fs (threshold %.3fs).", + lag, + warn_after, + ) + expected = now + interval + + +async def faulthandler_event_loop_watchdog( + *, + timeout: float = DEFAULT_WATCHDOG_TIMEOUT, + interval: float = DEFAULT_WATCHDOG_INTERVAL, + dump_file: TextIO | None = None, +) -> None: + """Dump all thread stacks if the event loop is blocked for too long. + + Args: + timeout: Seconds without watchdog refresh before faulthandler dumps stacks. + interval: Seconds between watchdog refreshes while the event loop is healthy. + dump_file: File object that receives faulthandler output. + """ + output = dump_file or sys.stderr + if not faulthandler.is_enabled(): + faulthandler.enable(file=output) + + try: + while True: + faulthandler.cancel_dump_traceback_later() + faulthandler.dump_traceback_later( + timeout, + repeat=False, + file=output, + ) + await asyncio.sleep(interval) + finally: + faulthandler.cancel_dump_traceback_later() + + +def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]: + """Create enabled event loop diagnostic tasks for the current loop. + + Returns: + A list of created asyncio tasks. + """ + settings = load_event_loop_diagnostic_settings() + tasks: list[asyncio.Task] = [] + + if settings.lag_monitor_enabled: + tasks.append( + asyncio.create_task( + monitor_event_loop_lag( + interval=settings.lag_monitor_interval, + warn_after=settings.lag_monitor_threshold, + ), + name="event_loop_lag_monitor", + ) + ) + + if settings.watchdog_enabled: + logger.warning( + "Event loop faulthandler watchdog enabled: timeout=%.3fs interval=%.3fs. " + "If the loop is blocked, Python thread stacks will be written to stderr.", + settings.watchdog_timeout, + settings.watchdog_interval, + ) + tasks.append( + asyncio.create_task( + faulthandler_event_loop_watchdog( + timeout=settings.watchdog_timeout, + interval=settings.watchdog_interval, + ), + name="event_loop_faulthandler_watchdog", + ) + ) + + return tasks diff --git a/tests/unit/test_event_loop_diagnostics.py b/tests/unit/test_event_loop_diagnostics.py new file mode 100644 index 000000000..a1e4d7144 --- /dev/null +++ b/tests/unit/test_event_loop_diagnostics.py @@ -0,0 +1,109 @@ +import asyncio + +import pytest + +from astrbot.core.utils import event_loop_diagnostics as diagnostics + + +def _clear_diagnostic_env(monkeypatch): + """Clear event loop diagnostic environment variables. + + Args: + monkeypatch: Pytest monkeypatch fixture. + """ + for name in ( + diagnostics.LAG_MONITOR_ENABLED_ENV, + diagnostics.LAG_MONITOR_INTERVAL_ENV, + diagnostics.LAG_MONITOR_THRESHOLD_ENV, + diagnostics.WATCHDOG_ENABLED_ENV, + diagnostics.WATCHDOG_INTERVAL_ENV, + diagnostics.WATCHDOG_TIMEOUT_ENV, + ): + monkeypatch.delenv(name, raising=False) + + +def test_load_event_loop_diagnostic_settings_defaults(monkeypatch): + """Default settings enable low-overhead lag monitoring only.""" + _clear_diagnostic_env(monkeypatch) + + settings = diagnostics.load_event_loop_diagnostic_settings() + + assert settings.lag_monitor_enabled is True + assert settings.lag_monitor_interval == diagnostics.DEFAULT_LAG_MONITOR_INTERVAL + assert settings.lag_monitor_threshold == diagnostics.DEFAULT_LAG_MONITOR_THRESHOLD + assert settings.watchdog_enabled is False + + +def test_load_event_loop_diagnostic_settings_from_env(monkeypatch): + """Environment variables override event loop diagnostic settings.""" + _clear_diagnostic_env(monkeypatch) + monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "0") + monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "yes") + monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "30") + + settings = diagnostics.load_event_loop_diagnostic_settings() + + assert settings.lag_monitor_enabled is False + assert settings.watchdog_enabled is True + assert settings.watchdog_timeout == 30 + + +@pytest.mark.asyncio +async def test_create_event_loop_diagnostic_tasks_respects_env(monkeypatch): + """Disabled lag monitor and watchdog should create no tasks.""" + _clear_diagnostic_env(monkeypatch) + monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "false") + monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "false") + + tasks = diagnostics.create_event_loop_diagnostic_tasks() + + assert tasks == [] + + +@pytest.mark.asyncio +async def test_create_event_loop_diagnostic_tasks_default_lag_monitor(monkeypatch): + """Default diagnostics should create only the lag monitor task.""" + _clear_diagnostic_env(monkeypatch) + + tasks = diagnostics.create_event_loop_diagnostic_tasks() + + try: + assert [task.get_name() for task in tasks] == ["event_loop_lag_monitor"] + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_faulthandler_watchdog_cancels_pending_dump(monkeypatch): + """The faulthandler watchdog should cancel its pending dump on shutdown.""" + calls = [] + + class FakeFaultHandler: + def is_enabled(self): + calls.append("is_enabled") + return False + + def enable(self, file): + calls.append(("enable", file)) + + def cancel_dump_traceback_later(self): + calls.append("cancel") + + def dump_traceback_later(self, timeout, repeat, file): + calls.append(("dump", timeout, repeat, file)) + + fake_faulthandler = FakeFaultHandler() + monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler) + + task = asyncio.create_task( + diagnostics.faulthandler_event_loop_watchdog(timeout=10, interval=1) + ) + await asyncio.sleep(0) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + assert "is_enabled" in calls + assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls) + assert calls[-1] == "cancel"