Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
221 changes: 169 additions & 52 deletions src/debug_toolbar/core/panels/async_profiler/detector.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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<func>[\w.]+)\(\).*?(?:running at|created at|defined at) (?P<file>.+?):(?P<line>\d+)"
)


def _get_stack_frames(
Expand Down Expand Up @@ -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.

Expand All @@ -58,10 +148,8 @@ class BlockingCallDetector:

__slots__ = (
"_blocking_calls",
"_log_handler",
"_loop",
"_original_debug",
"_original_duration",
"_original_handler",
"_threshold_ms",
)

Expand All @@ -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,
Expand Down Expand Up @@ -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.

Expand Down
8 changes: 7 additions & 1 deletion src/debug_toolbar/core/panels/async_profiler/taskfactory.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ class TaskFactoryBackend(AsyncProfilerBackend):
"_profiling_overhead",
"_start_time",
"_task_events",
"_task_refs",
)

def __init__(
Expand All @@ -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
Expand All @@ -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)

Expand Down Expand Up @@ -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))

Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading