diff --git a/src/debug_toolbar/core/panels/async_profiler/detector.py b/src/debug_toolbar/core/panels/async_profiler/detector.py index 6648aa7..a3b0465 100644 --- a/src/debug_toolbar/core/panels/async_profiler/detector.py +++ b/src/debug_toolbar/core/panels/async_profiler/detector.py @@ -4,16 +4,15 @@ import asyncio import logging +import re +import threading import time import traceback -from collections.abc import Callable -from typing import TYPE_CHECKING, Any +from typing import Any +from weakref import WeakKeyDictionary from debug_toolbar.core.panels.async_profiler.models import BlockingCall, LagSample -if TYPE_CHECKING: - ExceptionHandler = Callable[[asyncio.AbstractEventLoop, dict[str, Any]], object] - logger = logging.getLogger(__name__) DEFAULT_BLOCKING_THRESHOLD_MS = 100.0 @@ -22,6 +21,12 @@ DEFAULT_MAX_STACK_DEPTH = 10 STACK_SKIP_FRAMES = 3 MS_TO_SECONDS = 1000.0 +ASYNCIO_LOGGER_NAME = "asyncio" +SLOW_CALLBACK_LOG_ARGS = 2 + +_SLOW_CALLBACK_LOCATION_RE = re.compile( + r"(?P[\w.]+)\(\).*?(?:running at|created at|defined at) (?P.+?):(?P\d+)" +) def _get_stack_frames( @@ -49,6 +54,91 @@ def _get_stack_frames( return frames +class _LoopDebugPatchState: + """Tracks the refcounted patch of a loop's debug mode, slow callback + duration, and the ``asyncio`` logger level while one or more + ``BlockingCallDetector`` instances are installed on it. + """ + + __slots__ = ( + "logger_level_changed", + "original_debug", + "original_duration", + "original_logger_level", + "thresholds_ms", + ) + + def __init__(self) -> None: + self.original_debug: bool = False + self.original_duration: float = 0.1 + self.original_logger_level: int = logging.NOTSET + self.logger_level_changed: bool = False + self.thresholds_ms: dict[int, float] = {} + + +_loop_patch_lock = threading.Lock() +_loop_patch_states: WeakKeyDictionary[asyncio.AbstractEventLoop, _LoopDebugPatchState] = WeakKeyDictionary() + + +def _acquire_loop_patch(loop: asyncio.AbstractEventLoop, detector_key: int, threshold_ms: float) -> None: + """Enable debug mode and slow-callback logging on a loop, shared across detectors. + + Overlapping ``BlockingCallDetector.install()`` calls on the same loop share + one patch of the loop's debug state so the first detector to uninstall does + not disturb detectors still running, and the loop's original settings are + only restored once every detector has uninstalled. + + Args: + loop: The event loop to patch. + detector_key: A key unique to the calling detector (its ``id()``). + threshold_ms: The calling detector's blocking-call threshold. + """ + with _loop_patch_lock: + state = _loop_patch_states.get(loop) + if state is None: + state = _LoopDebugPatchState() + state.original_debug = loop.get_debug() + state.original_duration = loop.slow_callback_duration + asyncio_logger = logging.getLogger(ASYNCIO_LOGGER_NAME) + if asyncio_logger.getEffectiveLevel() > logging.WARNING: + state.original_logger_level = asyncio_logger.level + state.logger_level_changed = True + asyncio_logger.setLevel(logging.WARNING) + _loop_patch_states[loop] = state + + state.thresholds_ms[detector_key] = threshold_ms + loop.set_debug(True) + loop.slow_callback_duration = min(state.thresholds_ms.values()) / MS_TO_SECONDS + + +def _release_loop_patch(loop: asyncio.AbstractEventLoop, detector_key: int) -> None: + """Release one detector's share of a loop's debug patch. + + Restores the loop's original debug mode, slow callback duration, and + (if it was changed) the ``asyncio`` logger level once no detector remains + installed on the loop. + + Args: + loop: The event loop to release. + detector_key: The key passed to the matching ``_acquire_loop_patch`` call. + """ + with _loop_patch_lock: + state = _loop_patch_states.get(loop) + if state is None: + return + + state.thresholds_ms.pop(detector_key, None) + if state.thresholds_ms: + loop.slow_callback_duration = min(state.thresholds_ms.values()) / MS_TO_SECONDS + return + + loop.set_debug(state.original_debug) + loop.slow_callback_duration = state.original_duration + if state.logger_level_changed: + logging.getLogger(ASYNCIO_LOGGER_NAME).setLevel(state.original_logger_level) + del _loop_patch_states[loop] + + class BlockingCallDetector: """Detects blocking calls in async context. @@ -58,10 +148,8 @@ class BlockingCallDetector: __slots__ = ( "_blocking_calls", + "_log_handler", "_loop", - "_original_debug", - "_original_duration", - "_original_handler", "_threshold_ms", ) @@ -74,67 +162,41 @@ def __init__(self, threshold_ms: float = DEFAULT_BLOCKING_THRESHOLD_MS) -> None: self._threshold_ms = threshold_ms self._blocking_calls: list[BlockingCall] = [] self._loop: asyncio.AbstractEventLoop | None = None - self._original_debug: bool = False - self._original_duration: float = 0.1 - self._original_handler: ExceptionHandler | None = None + self._log_handler: _SlowCallbackLogHandler | None = None def install(self, loop: asyncio.AbstractEventLoop) -> None: """Install blocking call detection on the event loop. + Shares the loop's debug mode, slow callback duration, and ``asyncio`` + logger level with any other detector already installed on the same + loop, so overlapping requests can detect blocking calls concurrently + without one request's cleanup disturbing another's. + Args: loop: The asyncio event loop to monitor. """ self._loop = loop self._blocking_calls = [] - self._original_debug = loop.get_debug() - self._original_duration = loop.slow_callback_duration - self._original_handler = loop.get_exception_handler() + _acquire_loop_patch(loop, id(self), self._threshold_ms) - loop.set_debug(True) - loop.slow_callback_duration = self._threshold_ms / MS_TO_SECONDS - loop.set_exception_handler(self._exception_handler) + self._log_handler = _SlowCallbackLogHandler(self, self._threshold_ms) + logging.getLogger(ASYNCIO_LOGGER_NAME).addHandler(self._log_handler) - def _exception_handler(self, loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None: - """Custom exception handler that captures slow callback warnings. + def uninstall(self) -> None: + """Uninstall blocking call detection. - Args: - loop: The event loop. - context: Exception context dictionary. + Only restores the loop's debug mode, slow callback duration, and + logger level once every detector sharing the loop has uninstalled. """ - message = context.get("message", "") - if "Executing" in message and "took" in message and "seconds" in message: - import re - - match = re.search(r"took ([\d.]+) seconds", message) - if match: - duration_s = float(match.group(1)) - duration_ms = duration_s * MS_TO_SECONDS - source = context.get("source_traceback", []) - if source: - frame = source[-1] if source else None - self.record_blocking_call( - duration_ms=duration_ms, - function_name=frame.name if frame else "unknown", - file=frame.filename if frame else "unknown", - line=frame.lineno if frame else 0, - ) - else: - self.record_blocking_call(duration_ms=duration_ms) - - if self._original_handler: - self._original_handler(loop, context) - else: - loop.default_exception_handler(context) - - def uninstall(self) -> None: - """Uninstall blocking call detection and restore original settings.""" if self._loop is None: return - self._loop.set_debug(self._original_debug) - self._loop.slow_callback_duration = self._original_duration - self._loop.set_exception_handler(self._original_handler) + if self._log_handler is not None: + logging.getLogger(ASYNCIO_LOGGER_NAME).removeHandler(self._log_handler) + self._log_handler = None + + _release_loop_patch(self._loop, id(self)) def record_blocking_call( self, @@ -190,6 +252,61 @@ def get_stats(self) -> dict[str, Any]: } +class _SlowCallbackLogHandler(logging.Handler): + """Logging handler that turns asyncio's slow-callback warnings into blocking call records. + + CPython's event loop reports slow callbacks (when running in debug mode past + ``slow_callback_duration``) by calling ``logging.getLogger("asyncio").warning(...)`` + directly from ``BaseEventLoop._run_once``, not via ``call_exception_handler``. This + handler is attached directly to the ``asyncio`` logger to observe those records. + """ + + __slots__ = ("_detector", "_threshold_ms") + + def __init__(self, detector: BlockingCallDetector, threshold_ms: float) -> None: + super().__init__(level=logging.WARNING) + self._detector = detector + self._threshold_ms = threshold_ms + + def emit(self, record: logging.LogRecord) -> None: + """Parse a slow-callback warning record and record it as a blocking call. + + Args: + record: The log record emitted by the asyncio logger. + """ + if record.name != ASYNCIO_LOGGER_NAME or not isinstance(record.msg, str): + return + if not record.msg.startswith("Executing") or "took" not in record.msg: + return + + args = record.args + if not isinstance(args, tuple) or len(args) < SLOW_CALLBACK_LOG_ARGS: + return + + duration_arg = args[1] + if not isinstance(duration_arg, (int, float)): + return + duration_ms = float(duration_arg) * MS_TO_SECONDS + if duration_ms < self._threshold_ms: + return + + handle_repr = str(args[0]) + match = _SLOW_CALLBACK_LOCATION_RE.search(handle_repr) + if match: + function_name = match.group("func") + file = match.group("file") + line = int(match.group("line")) + else: + function_name, file, line = "unknown", "unknown", 0 + + self._detector.record_blocking_call( + duration_ms=duration_ms, + function_name=function_name, + file=file, + line=line, + ) + + class EventLoopLagMonitor: """Monitors event loop lag via scheduled callbacks. diff --git a/src/debug_toolbar/core/panels/async_profiler/taskfactory.py b/src/debug_toolbar/core/panels/async_profiler/taskfactory.py index dc1b915..f9a0846 100644 --- a/src/debug_toolbar/core/panels/async_profiler/taskfactory.py +++ b/src/debug_toolbar/core/panels/async_profiler/taskfactory.py @@ -69,6 +69,7 @@ class TaskFactoryBackend(AsyncProfilerBackend): "_profiling_overhead", "_start_time", "_task_events", + "_task_refs", ) def __init__( @@ -85,6 +86,7 @@ def __init__( """ self._original_factory: Any = None self._task_events: list[TaskEvent] = [] + self._task_refs: dict[str, asyncio.Task[Any]] = {} self._start_time: float = 0.0 self._loop: asyncio.AbstractEventLoop | None = None self._profiling_overhead: float = 0.0 @@ -102,6 +104,7 @@ def start(self, loop: asyncio.AbstractEventLoop) -> None: self._loop = loop self._start_time = loop.time() self._task_events = [] + self._task_refs = {} self._original_factory = loop.get_task_factory() loop.set_task_factory(self._profiling_task_factory) @@ -161,6 +164,7 @@ def _profiling_task_factory( stack_frames=stack_frames, ) self._task_events.append(event) + self._task_refs[event.task_id] = task task.add_done_callback(lambda t: self._record_task_complete(t, creation_time)) @@ -225,10 +229,12 @@ def get_stats(self) -> dict[str, Any]: """ tasks = [] for event in self._task_events: + ref = self._task_refs.get(event.task_id) + task_name = ref.get_name() if ref is not None else event.task_name tasks.append( { "task_id": event.task_id, - "task_name": event.task_name, + "task_name": task_name, "event_type": event.event_type, "timestamp": event.timestamp, "coro_name": event.coro_name, diff --git a/tests/unit/test_async_profiler_detector.py b/tests/unit/test_async_profiler_detector.py new file mode 100644 index 0000000..193bc36 --- /dev/null +++ b/tests/unit/test_async_profiler_detector.py @@ -0,0 +1,143 @@ +"""Tests for the async profiler blocking call detector.""" + +from __future__ import annotations + +import asyncio +import logging +import time + +from debug_toolbar.core.panels.async_profiler.detector import BlockingCallDetector + + +async def _blocking_handler() -> None: + time.sleep(0.15) # noqa: ASYNC251 -- intentional blocking call under test + + +async def test_detects_real_blocking_call() -> None: + """asyncio reports slow callbacks via the ``asyncio`` logger, not + ``call_exception_handler``, so the detector must hook the logger to see them. + """ + loop = asyncio.get_running_loop() + detector = BlockingCallDetector(threshold_ms=50) + detector.install(loop) + try: + task = loop.create_task(_blocking_handler()) + await asyncio.sleep(0.3) + await task + finally: + detector.uninstall() + + stats = detector.get_stats() + assert len(stats["blocking_calls"]) == 1 + call = stats["blocking_calls"][0] + assert call["function_name"] == "_blocking_handler" + assert call["duration_ms"] >= 100 + + +async def test_parses_windows_path_in_slow_callback_warning() -> None: + loop = asyncio.get_running_loop() + detector = BlockingCallDetector(threshold_ms=50) + detector.install(loop) + try: + logging.getLogger("asyncio").warning( + "Executing %s took %.3f seconds", + r">", + 0.2, + ) + finally: + detector.uninstall() + + call = detector.get_stats()["blocking_calls"][0] + assert call["function_name"] == "_blocking_handler" + assert call["file"] == r"C:\app\handlers.py" + assert call["line"] == 42 + + +async def test_no_false_positives_for_fast_code() -> None: + loop = asyncio.get_running_loop() + detector = BlockingCallDetector(threshold_ms=100) + detector.install(loop) + try: + await asyncio.sleep(0.01) + finally: + detector.uninstall() + + stats = detector.get_stats() + assert stats["blocking_calls"] == [] + + +async def test_uninstall_restores_logger_and_loop_state() -> None: + loop = asyncio.get_running_loop() + asyncio_logger = logging.getLogger("asyncio") + original_debug = loop.get_debug() + original_duration = loop.slow_callback_duration + original_level = asyncio_logger.level + original_handlers = list(asyncio_logger.handlers) + + detector = BlockingCallDetector(threshold_ms=50) + detector.install(loop) + detector.uninstall() + + assert loop.get_debug() == original_debug + assert loop.slow_callback_duration == original_duration + assert asyncio_logger.level == original_level + assert asyncio_logger.handlers == original_handlers + + +async def test_overlapping_detectors_share_loop_state_safely() -> None: + """Two detectors installed on the same loop at once must not clobber each + other's loop debug state: uninstalling one must not disturb the other, and + the loop's original settings are restored only once both have uninstalled. + """ + loop = asyncio.get_running_loop() + asyncio_logger = logging.getLogger("asyncio") + original_debug = loop.get_debug() + original_duration = loop.slow_callback_duration + original_level = asyncio_logger.level + + detector_a = BlockingCallDetector(threshold_ms=50) + detector_b = BlockingCallDetector(threshold_ms=80) + detector_a.install(loop) + detector_b.install(loop) + try: + task = loop.create_task(_blocking_handler()) + await asyncio.sleep(0.3) + await task + finally: + detector_a.uninstall() + assert loop.get_debug() is True + detector_b.uninstall() + + assert loop.get_debug() == original_debug + assert loop.slow_callback_duration == original_duration + assert asyncio_logger.level == original_level + + assert len(detector_a.get_stats()["blocking_calls"]) == 1 + assert len(detector_b.get_stats()["blocking_calls"]) == 1 + + +async def test_install_only_raises_logger_level_when_more_restrictive() -> None: + """A user's asyncio logger level more verbose than WARNING must survive + install/uninstall unchanged, while a level that would hide WARNING + records (e.g. ERROR) must be raised to WARNING while installed. + """ + loop = asyncio.get_running_loop() + asyncio_logger = logging.getLogger("asyncio") + + original_level = asyncio_logger.level + try: + asyncio_logger.setLevel(logging.DEBUG) + detector = BlockingCallDetector(threshold_ms=50) + detector.install(loop) + assert asyncio_logger.level == logging.DEBUG + detector.uninstall() + assert asyncio_logger.level == logging.DEBUG + + asyncio_logger.setLevel(logging.ERROR) + detector = BlockingCallDetector(threshold_ms=50) + detector.install(loop) + assert asyncio_logger.level == logging.WARNING + detector.uninstall() + assert asyncio_logger.level == logging.ERROR + finally: + asyncio_logger.setLevel(original_level) diff --git a/tests/unit/test_async_profiler_taskfactory.py b/tests/unit/test_async_profiler_taskfactory.py index 6200d8b..8e21f2d 100644 --- a/tests/unit/test_async_profiler_taskfactory.py +++ b/tests/unit/test_async_profiler_taskfactory.py @@ -25,3 +25,22 @@ async def test_create_task_with_eager_start_while_profiling() -> None: backend.stop() assert result == "ok" + + +async def test_task_name_reflects_create_task_name() -> None: + """asyncio applies the caller-supplied name after the task factory returns, + so the backend must resolve the real name lazily instead of capturing it + inside the factory call. + """ + loop = asyncio.get_running_loop() + backend = TaskFactoryBackend() + backend.start(loop) + try: + task = loop.create_task(_child(), name="my-named-task") + await task + finally: + backend.stop() + + stats = backend.get_stats() + created_names = [t["task_name"] for t in stats["tasks"] if t["event_type"] == "created"] + assert created_names == ["my-named-task"]