feat: detect event loop blocking in development

docs/advanced_onboarding/configuration.md

@@ -25,6 +25,19 @@ config = rx.Config(
 
 See the [config reference](/docs/api-reference/config/) for all the parameters available.
 
+## Event Loop Blocking Detection
+
+In development, Reflex warns when an event handler blocks the asyncio event loop near or beyond `event_loop_blocking_threshold` seconds (default `0.2`). The warning appears as soon as the threshold is reached, so a handler that never returns is reported too. It names the handler and the innermost line of your app's code in the blocking stack, then shows the full stack. A blocking task that processes no event, such as a task your app starts, is reported by its task name. When the loop responds again, a second line reports the total duration, which includes the state updates after the handler, such as computed vars. It recommends moving blocking I/O to a thread pool or using a non-blocking async API. Background events still run on the event loop and do not make synchronous blocking code non-blocking. Detection is always off in production. Set the threshold to `0` to turn it off in development:
+
+```python
+config = rx.Config(
+    app_name="my_app_name",
+    event_loop_blocking_threshold=0,
+)
+```
+
+The setting can also be overridden with `REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD`, for example `REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD=0` to turn detection off.
+
 ## Environment Variables
 
 Any config parameter can be overridden by setting an environment variable with the `REFLEX_` prefix and the parameter name in uppercase. Environment variables take precedence over values set in `rxconfig.py`.

news/+7619.feature.md

@@ -0,0 +1 @@
+Add development-time detection for event handlers that block the asyncio event loop.

packages/reflex-base/news/+7619.feature.md

@@ -0,0 +1 @@
+Add configurable development-time warnings for event handlers that block the asyncio event loop.

packages/reflex-base/src/reflex_base/config.py

@@ -3,6 +3,7 @@
 import dataclasses
 import importlib
 import logging
+import math
 import os
 import sys
 import threading
@@ -194,6 +195,7 @@ class BaseConfig:
         plugins: List of plugins to use in the app.
         disable_plugins: List of plugin types to disable in the app.
         transport: The transport method for client-server communication.
+        event_loop_blocking_threshold: Minimum event-loop blocking duration in seconds to report in development. Set to 0 to disable detection.
     """
 
     app_name: str
@@ -285,6 +287,8 @@ class BaseConfig:
 
     transport: Literal["websocket", "polling"] = "websocket"
 
+    event_loop_blocking_threshold: float = 0.2
+
     # Whether to skip plugin checks.
     _skip_plugins_checks: bool = dataclasses.field(default=False, repr=False)
 
@@ -399,6 +403,15 @@ def _post_init(self, **kwargs):
             msg = "trusted_proxy_hops must be a non-negative integer."
             raise ConfigError(msg)
 
+        threshold = self.event_loop_blocking_threshold
+        if (
+            isinstance(threshold, bool)
+            or not isinstance(threshold, (int, float))
+            or not 0 <= threshold < math.inf
+        ):
+            msg = "event_loop_blocking_threshold must be a non-negative finite number."
+            raise ConfigError(msg)
+
         # Normalize route prefixes to ensure they start with a slash.
         self._normalize_paths()
 

packages/reflex-base/src/reflex_base/event/processor/blocking.py

@@ -0,0 +1,186 @@
+"""Development-time detection of tasks blocking the asyncio loop."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import sys
+import sysconfig
+import threading
+import time
+import traceback
+from collections.abc import Mapping
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from reflex_base.registry import RegisteredEventHandler
+
+logger = logging.getLogger(__name__)
+
+# Name prefix of the asyncio task that processes one event.
+EVENT_TASK_PREFIX = "reflex_event|"
+
+# The standard library and installed packages hold no app code.
+_LIBRARY_PATHS = tuple({
+    f"{Path(sysconfig.get_path(name))}{os.sep}"
+    for name in ("stdlib", "platstdlib", "purelib", "platlib")
+})
+
+
+class EventLoopBlockingMonitor:
+    """Report tasks, such as event handlers, that keep the event loop from responding."""
+
+    def __init__(
+        self,
+        loop: asyncio.AbstractEventLoop,
+        threshold: float,
+        event_handlers: Mapping[str, RegisteredEventHandler],
+    ) -> None:
+        """Create a monitor for an event loop.
+
+        Args:
+            loop: The event loop to monitor.
+            threshold: Blocking duration in seconds at which to warn.
+            event_handlers: The registered handlers, by event name.
+        """
+        self.loop = loop
+        self.threshold = threshold
+        self.event_handlers = event_handlers
+        # The monitor is created on the loop thread, in the app root.
+        self._loop_thread_id = threading.get_ident()
+        self._app_root = f"{Path.cwd()}{os.sep}"
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+
+    def start(self) -> None:
+        """Start the single monitor thread for this event processor."""
+        self._thread = threading.Thread(
+            target=self._run,
+            name="reflex-event-loop-monitor",
+            daemon=True,
+        )
+        self._thread.start()
+
+    def stop(self) -> None:
+        """Stop the monitor thread."""
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join(timeout=1)
+            self._thread = None
+
+    def _describe_blocking(self) -> tuple[str, str, str, str] | None:
+        """Describe the task that runs on the blocked loop.
+
+        Returns:
+            The subject (event handler or task name), the scope note for event
+            handlers (or an empty string), the location of the innermost app
+            frame (or an empty string), and the formatted stack of the task, or
+            None if no task runs or the task has left the loop.
+        """
+        task = asyncio.current_task(self.loop)
+        if task is None or (stack := self._task_stack(task)) is None:
+            return None
+        name = task.get_name()
+        if name.startswith(EVENT_TASK_PREFIX):
+            event_name = name[len(EVENT_TASK_PREFIX) :].partition("|")[0]
+            registered = self.event_handlers.get(event_name)
+            handler_name = (
+                event_name if registered is None else registered.handler.fn.__qualname__
+            )
+            subject = f"Event handler {handler_name}"
+            scope = ", including state updates such as computed vars"
+        else:
+            subject, scope = f"Task {name!r}", ""
+        app_frame = self._app_frame(stack)
+        location = (
+            ""
+            if app_frame is None
+            else f" at {app_frame.filename[len(self._app_root) :]}:"
+            f"{app_frame.lineno} in {app_frame.name}"
+        )
+        return subject, scope, location, "".join(traceback.format_list(stack))
+
+    def _app_frame(
+        self, stack: traceback.StackSummary
+    ) -> traceback.FrameSummary | None:
+        """Find the innermost frame of app code, outside the stdlib and packages.
+
+        Args:
+            stack: The frames, outermost first.
+
+        Returns:
+            The innermost app frame, or None if no frame is app code.
+        """
+        for frame in reversed(stack):
+            if frame.filename.startswith(
+                self._app_root
+            ) and not frame.filename.startswith(_LIBRARY_PATHS):
+                return frame
+        return None
+
+    def _task_stack(self, task: asyncio.Task) -> traceback.StackSummary | None:
+        """Extract the loop thread's frames from the task entry to the blocking line.
+
+        Args:
+            task: The task that runs on the blocked loop.
+
+        Returns:
+            The frames, outermost first, or None if the loop thread no longer
+            runs the task.
+        """
+        entry = getattr(task.get_coro(), "cr_frame", None)
+        frame = sys._current_frames().get(self._loop_thread_id)
+        frames = []
+        while frame is not None:
+            frames.append(frame)
+            if frame is entry:
+                return traceback.StackSummary.extract(
+                    (frame, frame.f_lineno) for frame in reversed(frames)
+                )
+            frame = frame.f_back
+        return None
+
+    def _run(self) -> None:
+        """Poll loop responsiveness from outside the event-loop thread."""
+        # Poll at one tenth of the threshold, between one and 10 milliseconds.
+        interval = min(max(self.threshold / 10, 0.001), 0.01)
+        while not self._stop.wait(interval):
+            responsive = threading.Event()
+            started = time.perf_counter()
+            try:
+                self.loop.call_soon_threadsafe(responsive.set)
+            except RuntimeError:
+                return
+
+            # The task is only visible while the loop is blocked, so report it
+            # then, and report the total duration once the loop responds.
+            subject = None
+            while not responsive.wait(interval):
+                if self._stop.is_set():
+                    return
+                elapsed = time.perf_counter() - started
+                if (
+                    subject is None
+                    and elapsed + interval >= self.threshold
+                    and (blocking := self._describe_blocking()) is not None
+                ):
+                    subject, scope, location, stack = blocking
+                    logger.warning(
+                        "%s is blocking the event loop%s (threshold: %.3fs)%s. "
+                        "Move blocking I/O to a thread pool or use a non-blocking "
+                        "async API.\n"
+                        "Blocking stack (most recent call last):\n%s",
+                        subject,
+                        location,
+                        self.threshold,
+                        scope,
+                        stack.rstrip(),
+                    )
+            if subject is not None:
+                logger.warning(
+                    "%s blocked the event loop for %.3fs in total.",
+                    subject,
+                    time.perf_counter() - started,
+                )

packages/reflex-base/src/reflex_base/event/processor/event_processor.py

@@ -17,8 +17,13 @@
 
 from reflex.app_mixins.middleware import MiddlewareMixin
 from reflex.istate.manager import StateManager
-from reflex_base import otel
+from reflex_base import constants, otel
+from reflex_base.environment import environment
 from reflex_base.event.context import EventContext
+from reflex_base.event.processor.blocking import (
+    EVENT_TASK_PREFIX,
+    EventLoopBlockingMonitor,
+)
 from reflex_base.event.processor.future import EventFuture
 from reflex_base.event.processor.timeout import DrainTimeoutManager
 from reflex_base.registry import RegisteredEventHandler, RegistrationContext
@@ -108,6 +113,7 @@ class EventProcessor:
         Callable[[Exception], EventSpec | list[EventSpec] | None] | None
     ) = None
     graceful_shutdown_timeout: float | None = None
+    event_loop_blocking_threshold: float = 0
 
     _queue: asyncio.Queue[EventQueueEntry] | None = dataclasses.field(
         default=None, init=False
@@ -137,6 +143,9 @@ class EventProcessor:
         str,
         collections.deque[tuple[EventQueueEntry, RegisteredEventHandler]],
     ] = dataclasses.field(default_factory=dict, init=False)
+    _blocking_monitor: EventLoopBlockingMonitor | None = dataclasses.field(
+        default=None, init=False, repr=False
+    )
 
     def configure(
         self,
@@ -243,6 +252,16 @@ async def start(self) -> None:
             raise RuntimeError(msg)
         self._attached_root_context_token = EventContext.set(self._root_context)
         self._queue = asyncio.Queue()
+        if (
+            self.event_loop_blocking_threshold
+            and environment.REFLEX_ENV_MODE.get() == constants.Env.DEV
+        ):
+            self._blocking_monitor = EventLoopBlockingMonitor(
+                asyncio.get_running_loop(),
+                self.event_loop_blocking_threshold,
+                RegistrationContext.get().event_handlers,
+            )
+            self._blocking_monitor.start()
         self._ensure_queue_task()
 
     async def _stop_tasks(self, timeout: float | None = None) -> None:
@@ -339,6 +358,9 @@ async def stop(self, graceful_shutdown_timeout: float | None = None) -> None:
             if not future.done():
                 future.cancel()
         self._futures.clear()
+        if self._blocking_monitor is not None:
+            self._blocking_monitor.stop()
+            self._blocking_monitor = None
 
     async def join(
         self, timeout: float | None = None, queue: asyncio.Queue | None = None
@@ -772,7 +794,9 @@ def _create_event_task(
         coro = self._process_event_queue_entry(
             entry=entry, registered_handler=registered_handler
         )
-        name = f"reflex_event|{entry.event.name}|{entry.ctx.token}|{entry.ctx.txid}"
+        name = (
+            f"{EVENT_TASK_PREFIX}{entry.event.name}|{entry.ctx.token}|{entry.ctx.txid}"
+        )
         loop = asyncio.get_running_loop()
         if (
             sys.version_info >= (3, 12)

pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
   "starlette >=1.3.1",
   "typing_extensions >=4.13.0",
   "wrapt >=1.17.0,<2.6",
-  "reflex-base >= 0.10.1",
+  "reflex-base >= 0.10.2.dev0",
   "reflex-components-code >= 0.10.1",
   "reflex-components-core >= 0.10.0",
   "reflex-components-dataeditor >= 0.10.1",

reflex/app.py

@@ -768,7 +768,9 @@ async def _setup_event_processor(self) -> AsyncIterator[None]:
             event_namespace._token_manager._reset_instance_id()
         # Create the event processor.
         self._event_processor = BaseStateEventProcessor(
-            middleware=self, backend_exception_handler=self.backend_exception_handler
+            middleware=self,
+            backend_exception_handler=self.backend_exception_handler,
+            event_loop_blocking_threshold=get_config().event_loop_blocking_threshold,
         )
         async with self._event_processor.configure(
             state_manager=self.state_manager,