opensource/AstrBot
3
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot.git synced 2026-10-16 15:06:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: isolate event loop watchdog stack dumps

Browse Source
This commit is contained in:
Soulter
2026-09-04 00:34:10 +08:00
parent 077d54baaf
commit 7b49d92fff
2 changed files with 112 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
astrbot/core/utils/event_loop_diagnostics.py
+68 -24
View File
@@ -1,5 +1,8 @@
import asyncio
import faulthandler
import sys
import threading
import time
import traceback
from dataclasses import dataclass
from pathlib import Path
from typing import TextIO
@@ -25,10 +28,10 @@ class EventLoopDiagnosticSettings:
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_enabled: Whether to run the event loop watchdog.
watchdog_interval: Seconds between watchdog heartbeat refreshes.
watchdog_timeout: Seconds without event loop progress before dumping stacks.
watchdog_log_path: File that receives faulthandler watchdog output.
watchdog_log_path: File that receives watchdog stack dumps.
watchdog_log_max_bytes: Maximum watchdog log bytes before rotation.
"""
@@ -121,14 +124,14 @@ def _open_watchdog_log_file(log_path: Path, max_bytes: int) -> TextIO:
max_bytes: Maximum current log size before rotation.
Returns:
Writable text file object for faulthandler output.
Writable text file object for watchdog output.
"""
log_path.parent.mkdir(parents=True, exist_ok=True)
_rotate_watchdog_log_file(log_path, max_bytes)
return log_path.open("a", encoding="utf-8")
async def faulthandler_event_loop_watchdog(
async def event_loop_watchdog(
*,
timeout: float = DEFAULT_WATCHDOG_TIMEOUT,
interval: float = DEFAULT_WATCHDOG_INTERVAL,
@@ -139,36 +142,77 @@ async def faulthandler_event_loop_watchdog(
"""Dump all thread stacks if the event loop is blocked for too long.
Args:
timeout: Seconds without watchdog refresh before faulthandler dumps stacks.
timeout: Seconds without a heartbeat before the watchdog dumps stacks.
interval: Seconds between watchdog refreshes while the event loop is healthy.
dump_file: File object that receives faulthandler output.
dump_path: Path that receives faulthandler output when dump_file is unset.
dump_file: File object that receives stack dump output.
dump_path: Path that receives stack dumps when dump_file is unset.
max_bytes: Maximum current log size before rotation.
"""
log_path = dump_path or _watchdog_log_path()
try:
while True:
faulthandler.cancel_dump_traceback_later()
event_loop_thread_id = threading.get_ident()
stop_event = threading.Event()
heartbeat = time.monotonic()
def watch_heartbeat() -> None:
nonlocal heartbeat
dumped_heartbeat: float | None = None
while not stop_event.wait(interval):
observed_heartbeat = heartbeat
if (
observed_heartbeat == dumped_heartbeat
or time.monotonic() - observed_heartbeat < timeout
):
continue
output: TextIO | None = None
should_close = False
try:
output = dump_file or _open_watchdog_log_file(log_path, max_bytes)
should_close = dump_file is None
faulthandler.dump_traceback_later(
timeout,
repeat=False,
file=output,
stalled_for = time.monotonic() - observed_heartbeat
output.write(f"Event loop stalled for {stalled_for:.3f}s\n")
frames = sys._current_frames()
thread_names = {
thread.ident: thread.name for thread in threading.enumerate()
}
thread_ids = [event_loop_thread_id]
thread_ids.extend(
thread_id
for thread_id in frames
if thread_id != event_loop_thread_id
)
await asyncio.sleep(interval)
for thread_id in thread_ids:
frame = frames.get(thread_id)
if frame is None:
continue
thread_name = thread_names.get(thread_id, "unknown")
output.write(f'\nThread {thread_id} "{thread_name}"\n')
traceback.print_stack(frame, file=output)
output.flush()
dumped_heartbeat = observed_heartbeat
except Exception as e:
logger.warning("Event loop faulthandler watchdog failed: %s", e)
await asyncio.sleep(interval)
logger.warning("Event loop watchdog failed: %s", e)
finally:
faulthandler.cancel_dump_traceback_later()
if should_close and output is not None:
output.close()
watchdog_thread = threading.Thread(
target=watch_heartbeat,
name="event_loop_watchdog",
daemon=True,
)
watchdog_thread.start()
try:
while True:
await asyncio.sleep(interval)
heartbeat = time.monotonic()
finally:
faulthandler.cancel_dump_traceback_later()
stop_event.set()
watchdog_thread.join(timeout=interval + 1)
if watchdog_thread.is_alive():
logger.warning("Event loop watchdog thread did not stop cleanly.")
def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]:
@@ -193,7 +237,7 @@ def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]:
if settings.watchdog_enabled:
logger.info(
"Event loop faulthandler watchdog enabled: timeout=%.3fs interval=%.3fs. "
"Event loop watchdog enabled: timeout=%.3fs interval=%.3fs. "
"If the loop is blocked, Python thread stacks will be written to %s "
"(rotates at %d bytes).",
settings.watchdog_timeout,
@@ -203,13 +247,13 @@ def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]:
)
tasks.append(
asyncio.create_task(
faulthandler_event_loop_watchdog(
event_loop_watchdog(
timeout=settings.watchdog_timeout,
interval=settings.watchdog_interval,
dump_path=settings.watchdog_log_path,
max_bytes=settings.watchdog_log_max_bytes,
),
name="event_loop_faulthandler_watchdog",
name="event_loop_watchdog",
)
)
tests/unit/test_event_loop_diagnostics.py
+44 -64
View File
@@ -1,4 +1,6 @@
import asyncio
import threading
import time
import pytest
@@ -6,7 +8,7 @@ from astrbot.core.utils import event_loop_diagnostics as diagnostics
def test_load_event_loop_diagnostic_settings_defaults():
"""Default settings enable lag monitoring and stack dump watchdog."""
"""Default settings enable lag monitoring and the stack dump watchdog."""
settings = diagnostics.load_event_loop_diagnostic_settings()
assert settings.lag_monitor_enabled is True
@@ -26,7 +28,7 @@ async def test_create_event_loop_diagnostic_tasks_defaults():
try:
assert [task.get_name() for task in tasks] == [
"event_loop_lag_monitor",
"event_loop_faulthandler_watchdog",
"event_loop_watchdog",
]
finally:
for task in tasks:
@@ -35,100 +37,78 @@ async def test_create_event_loop_diagnostic_tasks_defaults():
@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 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)
async def test_event_loop_watchdog_stops_worker_thread():
"""The event loop watchdog should stop its worker thread on shutdown."""
task = asyncio.create_task(
diagnostics.faulthandler_event_loop_watchdog(timeout=10, interval=1)
diagnostics.event_loop_watchdog(timeout=10, interval=0.01)
)
await asyncio.sleep(0)
await asyncio.sleep(0.02)
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls)
assert calls[-1] == "cancel"
assert not any(
thread.name == "event_loop_watchdog" for thread in threading.enumerate()
)
@pytest.mark.asyncio
async def test_faulthandler_watchdog_writes_rotating_log(tmp_path, monkeypatch):
"""The faulthandler watchdog should write to and rotate its log file."""
async def test_event_loop_watchdog_writes_rotating_log(tmp_path):
"""The watchdog should write to and rotate its log file."""
log_path = tmp_path / "logs" / "event_loop_watchdog.log"
log_path.parent.mkdir()
log_path.write_text("x" * 8, encoding="utf-8")
calls = []
class FakeFaultHandler:
def cancel_dump_traceback_later(self):
calls.append("cancel")
def dump_traceback_later(self, timeout, repeat, file):
calls.append(("dump", timeout, repeat, file.name))
file.write("watchdog dump\n")
file.flush()
fake_faulthandler = FakeFaultHandler()
monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler)
task = asyncio.create_task(
diagnostics.faulthandler_event_loop_watchdog(
timeout=10,
interval=1,
diagnostics.event_loop_watchdog(
timeout=0.02,
interval=0.005,
dump_path=log_path,
max_bytes=4,
)
)
await asyncio.sleep(0)
time.sleep(0.05) # noqa: ASYNC251 - Intentionally block the event loop.
await asyncio.sleep(0.02)
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert log_path.read_text(encoding="utf-8") == "watchdog dump\n"
assert log_path.with_name("event_loop_watchdog.log.1").read_text(
encoding="utf-8"
) == "x" * 8
assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls)
log_content = log_path.read_text(encoding="utf-8")
assert "Event loop stalled for" in log_content
assert "test_event_loop_diagnostics.py" in log_content
assert (
log_path.with_name("event_loop_watchdog.log.1").read_text(encoding="utf-8")
== "x" * 8
)
@pytest.mark.asyncio
async def test_faulthandler_watchdog_survives_dump_failure(tmp_path, monkeypatch):
"""The watchdog should keep running after faulthandler arm failures."""
async def test_event_loop_watchdog_survives_dump_failure(tmp_path, monkeypatch):
"""The watchdog should keep running after stack dump failures."""
log_path = tmp_path / "event_loop_watchdog.log"
armed_again = asyncio.Event()
calls = []
dumped = threading.Event()
attempts = 0
class FakeFaultHandler:
def cancel_dump_traceback_later(self):
calls.append("cancel")
def flaky_open(path, max_bytes):
nonlocal attempts
attempts += 1
if attempts == 1:
raise OSError("boom")
dumped.set()
return path.open("a", encoding="utf-8")
def dump_traceback_later(self, timeout, repeat, file):
calls.append(("dump", timeout, repeat, file.name))
if len([call for call in calls if isinstance(call, tuple)]) == 1:
raise RuntimeError("boom")
armed_again.set()
fake_faulthandler = FakeFaultHandler()
monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler)
monkeypatch.setattr(diagnostics, "_open_watchdog_log_file", flaky_open)
task = asyncio.create_task(
diagnostics.faulthandler_event_loop_watchdog(
timeout=10,
interval=0.01,
diagnostics.event_loop_watchdog(
timeout=0.02,
interval=0.005,
dump_path=log_path,
)
)
await asyncio.wait_for(armed_again.wait(), timeout=1)
await asyncio.sleep(0)
time.sleep(0.06) # noqa: ASYNC251 - Intentionally block the event loop.
assert dumped.is_set()
task.cancel()
await asyncio.gather(task, return_exceptions=True)
dump_calls = [call for call in calls if isinstance(call, tuple)]
assert len(dump_calls) >= 2
assert attempts >= 2