From 1349397c7c14ec51ed249e00f4aae1ff96b304fa Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 27 Oct 2026 19:53:50 +0530 Subject: [PATCH 01/12] feat: detect event loop blocking in development --- docs/advanced_onboarding/configuration.md | 13 ++ news/+7619.feature.md | 1 + .../reflex-base/src/reflex_base/config.py | 15 +++ .../reflex_base/event/processor/blocking.py | 114 ++++++++++++++++++ .../event/processor/event_processor.py | 45 +++++-- reflex/app.py | 4 +- .../event/processor/test_blocking.py | 72 +++++++++++ .../event/processor/test_event_processor.py | 51 +++++++- tests/units/test_config.py | 25 ++++ 9 files changed, 328 insertions(+), 12 deletions(-) create mode 100644 news/+7619.feature.md create mode 100644 packages/reflex-base/src/reflex_base/event/processor/blocking.py create mode 100644 tests/units/reflex_base/event/processor/test_blocking.py diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index 92d7e67567f..570d15e31f6 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/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 reports when an event handler blocks the asyncio event loop for at least `event_loop_blocking_threshold` seconds (default `0.2`). The warning includes the handler name and suggests moving blocking work to a thread pool executor or a background event. Detection is disabled in production. Set the threshold to `None` to disable it in development: + +```python +config = rx.Config( + app_name="my_app_name", + event_loop_blocking_threshold=None, +) +``` + +The setting can also be overridden with `REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD`. + ## 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`. diff --git a/news/+7619.feature.md b/news/+7619.feature.md new file mode 100644 index 00000000000..eeb9c33a28b --- /dev/null +++ b/news/+7619.feature.md @@ -0,0 +1 @@ +Add development-time detection for event handlers that block the asyncio event loop. diff --git a/packages/reflex-base/src/reflex_base/config.py b/packages/reflex-base/src/reflex_base/config.py index c3861dc4c3d..02e2329bfd8 100644 --- a/packages/reflex-base/src/reflex_base/config.py +++ b/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 None to disable detection. """ app_name: str @@ -285,6 +287,8 @@ class BaseConfig: transport: Literal["websocket", "polling"] = "websocket" + event_loop_blocking_threshold: float | None = 0.2 + # Whether to skip plugin checks. _skip_plugins_checks: bool = dataclasses.field(default=False, repr=False) @@ -399,6 +403,17 @@ def _post_init(self, **kwargs): msg = "trusted_proxy_hops must be a non-negative integer." raise ConfigError(msg) + if self.event_loop_blocking_threshold is not None and ( + isinstance(self.event_loop_blocking_threshold, bool) + or not isinstance(self.event_loop_blocking_threshold, (int, float)) + or not math.isfinite(self.event_loop_blocking_threshold) + or self.event_loop_blocking_threshold <= 0 + ): + msg = ( + "event_loop_blocking_threshold must be a positive finite number or None." + ) + raise ConfigError(msg) + # Normalize route prefixes to ensure they start with a slash. self._normalize_paths() diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py new file mode 100644 index 00000000000..0e6c4250b25 --- /dev/null +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -0,0 +1,114 @@ +"""Development-time detection of event handlers blocking the asyncio loop.""" + +from __future__ import annotations + +import asyncio +import logging +import sys +import threading +import time +import types + +logger = logging.getLogger(__name__) + + +class EventLoopBlockingMonitor: + """Report event handlers that keep the event loop from responding.""" + + def __init__(self, loop: asyncio.AbstractEventLoop, threshold: float) -> None: + self.loop = loop + self.threshold = threshold + self._loop_thread_id = threading.get_ident() + self._stop = threading.Event() + self._handlers_lock = threading.Lock() + self._handlers: dict[types.CodeType, tuple[str, int]] = {} + 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 handler_started(self, handler: object) -> None: + """Record a handler while its event task is executing. + + Args: + handler: The registered handler function. + """ + code = getattr(handler, "__code__", None) + name = getattr(handler, "__qualname__", None) + if not isinstance(code, types.CodeType) or not isinstance(name, str): + return + with self._handlers_lock: + current_name, count = self._handlers.get(code, (name, 0)) + self._handlers[code] = current_name, count + 1 + + def handler_finished(self, handler: object) -> None: + """Remove a handler after its event task finishes. + + Args: + handler: The registered handler function. + """ + code = getattr(handler, "__code__", None) + if not isinstance(code, types.CodeType): + return + with self._handlers_lock: + if (current := self._handlers.get(code)) is None: + return + name, count = current + if count == 1: + del self._handlers[code] + else: + self._handlers[code] = name, count - 1 + + def _find_handler(self) -> str | None: + """Find the innermost active handler frame on the event-loop thread.""" + frame = sys._current_frames().get(self._loop_thread_id) + if frame is None: + return None + with self._handlers_lock: + handlers = self._handlers.copy() + while frame is not None: + if handler := handlers.get(frame.f_code): + return handler[0] + frame = frame.f_back + return None + + def _run(self) -> None: + """Poll loop responsiveness from outside the event-loop thread.""" + interval = min(max(self.threshold / 2, 0.001), 0.1) + while not self._stop.wait(interval): + responsive = threading.Event() + started = time.perf_counter() + try: + self.loop.call_soon_threadsafe(responsive.set) + except RuntimeError: + return + + reported = False + while not responsive.wait(interval): + if self._stop.is_set(): + return + duration = time.perf_counter() - started + if duration >= self.threshold and not reported: + if handler_name := self._find_handler(): + logger.warning( + "Event handler %s blocked the event loop for at least " + "%.3fs (threshold: %.3fs). Move blocking work to a " + "thread pool executor or a background event.", + handler_name, + duration, + self.threshold, + ) + reported = True diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py index 51f7e1231be..ca7ede0dc5c 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py @@ -17,8 +17,10 @@ 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 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 +110,7 @@ class EventProcessor: Callable[[Exception], EventSpec | list[EventSpec] | None] | None ) = None graceful_shutdown_timeout: float | None = None + event_loop_blocking_threshold: float | None = None _queue: asyncio.Queue[EventQueueEntry] | None = dataclasses.field( default=None, init=False @@ -137,6 +140,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 +249,14 @@ 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 is not None + and environment.REFLEX_ENV_MODE.get() == constants.Env.DEV + ): + self._blocking_monitor = EventLoopBlockingMonitor( + asyncio.get_running_loop(), self.event_loop_blocking_threshold + ) + self._blocking_monitor.start() self._ensure_queue_task() async def _stop_tasks(self, timeout: float | None = None) -> None: @@ -339,6 +353,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 @@ -744,15 +761,23 @@ async def _process_event_queue_entry( """ # Set up the event context for this task. EventContext.set(entry.ctx) - if not otel.enabled: - await self._execute_event( - entry=entry, registered_handler=registered_handler - ) - return - with otel.event_span(entry.event, entry.ctx, registered_handler): - await self._execute_event( - entry=entry, registered_handler=registered_handler - ) + monitor = self._blocking_monitor + handler = registered_handler.handler.fn + if monitor is not None: + monitor.handler_started(handler) + try: + if not otel.enabled: + await self._execute_event( + entry=entry, registered_handler=registered_handler + ) + return + with otel.event_span(entry.event, entry.ctx, registered_handler): + await self._execute_event( + entry=entry, registered_handler=registered_handler + ) + finally: + if monitor is not None: + monitor.handler_finished(handler) def _create_event_task( self, diff --git a/reflex/app.py b/reflex/app.py index 97d7d7472e5..39ce47ec1c9 100644 --- a/reflex/app.py +++ b/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, diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py new file mode 100644 index 00000000000..426aab58a96 --- /dev/null +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -0,0 +1,72 @@ +"""Tests for development-time event-loop blocking detection.""" + +import asyncio +import logging +import time + +import pytest + +from reflex_base.event.processor.blocking import EventLoopBlockingMonitor + + +async def test_sync_handler_blocking_is_reported(caplog: pytest.LogCaptureFixture): + """Report the qualified name and duration of a blocking sync handler.""" + + def blocking_handler(): + time.sleep(0.12) + + monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + monitor.handler_started(blocking_handler) + monitor.start() + try: + blocking_handler() + await asyncio.sleep(0) + finally: + monitor.handler_finished(blocking_handler) + monitor.stop() + + assert "blocking_handler" in caplog.text + assert "blocked the event loop" in caplog.text + assert "thread pool executor or a background event" in caplog.text + + +async def test_async_handler_blocking_is_reported(caplog: pytest.LogCaptureFixture): + """Report synchronous blocking code inside an async handler.""" + + async def blocking_handler(): + time.sleep(0.12) + + monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + monitor.handler_started(blocking_handler) + monitor.start() + try: + await blocking_handler() + await asyncio.sleep(0) + finally: + monitor.handler_finished(blocking_handler) + monitor.stop() + + assert "blocking_handler" in caplog.text + + +async def test_async_handler_await_does_not_report_as_blocking( + caplog: pytest.LogCaptureFixture, +): + """Do not report handlers that yield control while doing async work.""" + + async def yielding_handler(): + await asyncio.sleep(0.1) + + monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + monitor.handler_started(yielding_handler) + monitor.start() + try: + await yielding_handler() + finally: + monitor.handler_finished(yielding_handler) + monitor.stop() + + assert "blocked the event loop" not in caplog.text diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index e7a389b4392..0ffca9b6e2c 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -5,6 +5,7 @@ import dataclasses import logging import sys +import time from collections.abc import AsyncIterator, Mapping from typing import Any from unittest.mock import Mock @@ -12,7 +13,7 @@ import pytest from opentelemetry.trace import SpanKind from pytest_mock import MockerFixture -from reflex_base import otel +from reflex_base import constants, otel from reflex_base.event.context import EventContext from reflex_base.event.processor.event_processor import ( EventProcessor, @@ -42,6 +43,11 @@ async def _slow_handler(delay: float = 0.5): await asyncio.sleep(delay) +def _blocking_sync_handler(): + """Block the event loop briefly for the detector integration test.""" + time.sleep(0.12) + + async def _error_handler(): # noqa: RUF029 """A handler that always raises.""" raise RuntimeError("boom") # noqa: EM101 @@ -364,6 +370,7 @@ async def _counting_superseding_handler(tick: int = 0, limit: int = 3): noop_event = EventHandler(fn=_noop_handler) slow_event = EventHandler(fn=_slow_handler) +blocking_sync_event = EventHandler(fn=_blocking_sync_handler) error_event = EventHandler(fn=_error_handler) recovery_event = EventHandler(fn=_recovery_handler) logging_event = EventHandler(fn=_logging_handler) @@ -404,6 +411,7 @@ def _register_handlers(forked_registration_context: RegistrationContext): for handler in ( noop_event, slow_event, + blocking_sync_event, error_event, recovery_event, logging_event, @@ -517,6 +525,27 @@ async def test_start_twice(processor: EventProcessor): await processor.stop() +async def test_blocking_monitor_starts_only_in_development( + processor: EventProcessor, monkeypatch: pytest.MonkeyPatch +): + """Enable the loop monitor in dev and leave it off in production.""" + processor.event_loop_blocking_threshold = 0.05 + processor.configure() + + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.PROD.value) + await processor.start() + assert processor._blocking_monitor is None + await processor.stop() + + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + dev_processor = EventProcessor(event_loop_blocking_threshold=0.05) + dev_processor.configure() + await dev_processor.start() + assert dev_processor._blocking_monitor is not None + await dev_processor.stop() + assert dev_processor._blocking_monitor is None + + async def test_stop_idempotent(processor: EventProcessor): """Stopping an already-stopped processor does not error. @@ -613,6 +642,26 @@ async def test_events_are_processed( assert _CALL_LOG == [{"value": "hello"}] +async def test_processor_reports_blocking_sync_handler( + mock_event_processor: EventProcessor, + token: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +): + """Report blocking handlers executed through the event queue.""" + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + mock_event_processor.event_loop_blocking_threshold = 0.04 + caplog.set_level( + logging.WARNING, logger="reflex_base.event.processor.blocking" + ) + + async with mock_event_processor as ep: + await ep.enqueue(token, Event.from_event_type(blocking_sync_event())[0]) + + assert "_blocking_sync_handler" in caplog.text + assert "blocked the event loop" in caplog.text + + async def test_enqueue_returns_future( mock_event_processor: EventProcessor, token: str, diff --git a/tests/units/test_config.py b/tests/units/test_config.py index 4ec4bc3d68c..00634456ad5 100644 --- a/tests/units/test_config.py +++ b/tests/units/test_config.py @@ -72,6 +72,31 @@ def test_trusted_proxy_hops_defaults_to_zero(base_config_values): assert rx.Config(**base_config_values).trusted_proxy_hops == 0 +def test_event_loop_blocking_threshold_defaults_to_development_threshold( + base_config_values, +): + """Enable useful blocking diagnostics by default in development.""" + assert rx.Config(**base_config_values).event_loop_blocking_threshold == 0.2 + + +@pytest.mark.parametrize("threshold", [0, -1, float("inf"), float("nan"), True, "1"]) +def test_event_loop_blocking_threshold_must_be_positive_finite_or_none( + base_config_values, threshold +): + """Reject invalid event-loop blocking thresholds.""" + with pytest.raises( + ConfigError, + match="event_loop_blocking_threshold must be a positive finite number or None", + ): + rx.Config(**base_config_values, event_loop_blocking_threshold=threshold) + + +def test_event_loop_blocking_detection_can_be_disabled(base_config_values): + """Allow apps to disable event-loop blocking detection.""" + config = rx.Config(**base_config_values, event_loop_blocking_threshold=None) + assert config.event_loop_blocking_threshold is None + + @pytest.mark.parametrize("trusted_proxy_hops", [-1, True, False, 1.5, "2", None]) def test_trusted_proxy_hops_must_be_non_negative( base_config_values, trusted_proxy_hops From 53b10f4dab35059ec97effae9039a9b9c282e0ab Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 27 Oct 2026 20:19:48 +0530 Subject: [PATCH 02/12] fix: address event loop monitor review feedback --- docs/advanced_onboarding/configuration.md | 2 +- packages/reflex-base/news/+7619.feature.md | 1 + .../reflex-base/src/reflex_base/config.py | 22 +++++++++------- .../reflex_base/event/processor/blocking.py | 11 ++++---- pyproject.toml | 2 +- .../event/processor/test_blocking.py | 25 ++++++++++++++++++- tests/units/test_app.py | 13 ++++++++++ tests/units/test_config.py | 5 +++- 8 files changed, 63 insertions(+), 18 deletions(-) create mode 100644 packages/reflex-base/news/+7619.feature.md diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index 570d15e31f6..a11fa37e3ee 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/docs/advanced_onboarding/configuration.md @@ -27,7 +27,7 @@ See the [config reference](/docs/api-reference/config/) for all the parameters a ## Event Loop Blocking Detection -In development, Reflex reports when an event handler blocks the asyncio event loop for at least `event_loop_blocking_threshold` seconds (default `0.2`). The warning includes the handler name and suggests moving blocking work to a thread pool executor or a background event. Detection is disabled in production. Set the threshold to `None` to disable it in development: +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 names the handler and 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 disabled in production. Set the threshold to `None` to disable it in development: ```python config = rx.Config( diff --git a/packages/reflex-base/news/+7619.feature.md b/packages/reflex-base/news/+7619.feature.md new file mode 100644 index 00000000000..8e05e63ca30 --- /dev/null +++ b/packages/reflex-base/news/+7619.feature.md @@ -0,0 +1 @@ +Add configurable development-time warnings for event handlers that block the asyncio event loop. diff --git a/packages/reflex-base/src/reflex_base/config.py b/packages/reflex-base/src/reflex_base/config.py index 02e2329bfd8..a374e909655 100644 --- a/packages/reflex-base/src/reflex_base/config.py +++ b/packages/reflex-base/src/reflex_base/config.py @@ -403,16 +403,20 @@ def _post_init(self, **kwargs): msg = "trusted_proxy_hops must be a non-negative integer." raise ConfigError(msg) - if self.event_loop_blocking_threshold is not None and ( - isinstance(self.event_loop_blocking_threshold, bool) - or not isinstance(self.event_loop_blocking_threshold, (int, float)) - or not math.isfinite(self.event_loop_blocking_threshold) - or self.event_loop_blocking_threshold <= 0 - ): - msg = ( - "event_loop_blocking_threshold must be a positive finite number or None." + if self.event_loop_blocking_threshold is not None: + threshold = self.event_loop_blocking_threshold + is_number = not isinstance(threshold, bool) and isinstance( + threshold, (int, float) ) - raise ConfigError(msg) + try: + is_finite = is_number and math.isfinite(threshold) + except OverflowError: + is_finite = False + if not is_finite or threshold <= 0: + msg = ( + "event_loop_blocking_threshold must be a positive finite number or None." + ) + raise ConfigError(msg) # Normalize route prefixes to ensure they start with a slash. self._normalize_paths() diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index 0e6c4250b25..62c8a8ac37e 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -87,7 +87,8 @@ def _find_handler(self) -> str | None: def _run(self) -> None: """Poll loop responsiveness from outside the event-loop thread.""" - interval = min(max(self.threshold / 2, 0.001), 0.1) + # 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() @@ -101,12 +102,12 @@ def _run(self) -> None: if self._stop.is_set(): return duration = time.perf_counter() - started - if duration >= self.threshold and not reported: + if duration + interval >= self.threshold and not reported: if handler_name := self._find_handler(): logger.warning( - "Event handler %s blocked the event loop for at least " - "%.3fs (threshold: %.3fs). Move blocking work to a " - "thread pool executor or a background event.", + "Event handler %s blocked the event loop " + "(observed delay: %.3fs; threshold: %.3fs). Move blocking " + "I/O to a thread pool or use a non-blocking async API.", handler_name, duration, self.threshold, diff --git a/pyproject.toml b/pyproject.toml index 8d59db5a773..1eef5b8d253 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index 426aab58a96..13167236b3d 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -28,7 +28,30 @@ def blocking_handler(): assert "blocking_handler" in caplog.text assert "blocked the event loop" in caplog.text - assert "thread pool executor or a background event" in caplog.text + assert "thread pool or use a non-blocking async API" in caplog.text + + +async def test_handler_blocking_just_over_threshold_is_detected( + caplog: pytest.LogCaptureFixture, +): + """Account for the time between responsiveness probes.""" + + def blocking_handler(): + time.sleep(0.055) + + monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + monitor.handler_started(blocking_handler) + monitor.start() + try: + await asyncio.sleep(0.021) + blocking_handler() + await asyncio.sleep(0) + finally: + monitor.handler_finished(blocking_handler) + monitor.stop() + + assert "blocking_handler" in caplog.text async def test_async_handler_blocking_is_reported(caplog: pytest.LogCaptureFixture): diff --git a/tests/units/test_app.py b/tests/units/test_app.py index ae575628094..61b60a09e45 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -3520,6 +3520,19 @@ def test_app_state_determination(): assert a2._state is None +async def test_setup_event_processor_passes_blocking_threshold( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture +): + """Forward the app config threshold to its event processor.""" + app = App() + config = rx.Config(app_name="blockingthreshold", event_loop_blocking_threshold=0.35) + mocker.patch("reflex.app.get_config", return_value=config) + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.PROD.value) + + async with app._setup_event_processor(): + assert app.event_processor.event_loop_blocking_threshold == 0.35 + + def test_raise_on_state(): """Test that the state is set.""" # state kwargs is deprecated, we just make sure the app is created anyway. diff --git a/tests/units/test_config.py b/tests/units/test_config.py index 00634456ad5..2e42c2ef430 100644 --- a/tests/units/test_config.py +++ b/tests/units/test_config.py @@ -79,7 +79,9 @@ def test_event_loop_blocking_threshold_defaults_to_development_threshold( assert rx.Config(**base_config_values).event_loop_blocking_threshold == 0.2 -@pytest.mark.parametrize("threshold", [0, -1, float("inf"), float("nan"), True, "1"]) +@pytest.mark.parametrize( + "threshold", [0, -1, float("inf"), float("nan"), True, "1", 10**1000] +) def test_event_loop_blocking_threshold_must_be_positive_finite_or_none( base_config_values, threshold ): @@ -139,6 +141,7 @@ def test_react_compiler_default(base_config_values): ("REFLEX_REACT_COMPILER", False), ("REFLEX_REACT_COMPILER", True), ("REFLEX_DEFAULT_COLOR_MODE", "dark"), + ("REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD", 0.35), ], ) def test_update_from_env( From 745b9999d61529a7bfd752331d6c5d8fda761f7a Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 27 Oct 2026 21:07:47 +0530 Subject: [PATCH 03/12] test: stabilize event loop blocking coverage --- .../reflex-base/src/reflex_base/config.py | 4 +--- .../reflex_base/event/processor/blocking.py | 6 +++++ .../event/processor/test_blocking.py | 24 +++++++++++++++++-- 3 files changed, 29 insertions(+), 5 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/config.py b/packages/reflex-base/src/reflex_base/config.py index a374e909655..2bc18e10a4e 100644 --- a/packages/reflex-base/src/reflex_base/config.py +++ b/packages/reflex-base/src/reflex_base/config.py @@ -413,9 +413,7 @@ def _post_init(self, **kwargs): except OverflowError: is_finite = False if not is_finite or threshold <= 0: - msg = ( - "event_loop_blocking_threshold must be a positive finite number or None." - ) + msg = "event_loop_blocking_threshold must be a positive finite number or None." raise ConfigError(msg) # Normalize route prefixes to ensure they start with a slash. diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index 62c8a8ac37e..b30a50a2b99 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -16,6 +16,12 @@ class EventLoopBlockingMonitor: """Report event handlers that keep the event loop from responding.""" def __init__(self, loop: asyncio.AbstractEventLoop, threshold: float) -> None: + """Create a monitor for an event loop. + + Args: + loop: The event loop to monitor. + threshold: Blocking duration in seconds at which to warn. + """ self.loop = loop self.threshold = threshold self._loop_thread_id = threading.get_ident() diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index 13167236b3d..7cd9280a432 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -1,8 +1,11 @@ """Tests for development-time event-loop blocking detection.""" +from collections.abc import Callable import asyncio import logging +import threading import time +from typing import Any import pytest @@ -20,6 +23,8 @@ def blocking_handler(): monitor.handler_started(blocking_handler) monitor.start() try: + # Let the monitor queue a responsiveness probe before blocking the loop. + await asyncio.sleep(0.02) blocking_handler() await asyncio.sleep(0) finally: @@ -33,18 +38,31 @@ def blocking_handler(): async def test_handler_blocking_just_over_threshold_is_detected( caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, ): """Account for the time between responsiveness probes.""" + loop = asyncio.get_running_loop() + probe_queued = threading.Event() + call_soon_threadsafe = loop.call_soon_threadsafe + + def track_probe( + callback: Callable[..., Any], *args: Any, **kwargs: Any + ) -> asyncio.Handle: + handle = call_soon_threadsafe(callback, *args, **kwargs) + probe_queued.set() + return handle + + monkeypatch.setattr(loop, "call_soon_threadsafe", track_probe) def blocking_handler(): + assert probe_queued.wait(timeout=1) time.sleep(0.055) - monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) + monitor = EventLoopBlockingMonitor(loop, threshold=0.04) caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") monitor.handler_started(blocking_handler) monitor.start() try: - await asyncio.sleep(0.021) blocking_handler() await asyncio.sleep(0) finally: @@ -65,6 +83,8 @@ async def blocking_handler(): monitor.handler_started(blocking_handler) monitor.start() try: + # Let the monitor queue a responsiveness probe before blocking the loop. + await asyncio.sleep(0.02) await blocking_handler() await asyncio.sleep(0) finally: From 12c49640a5e0c58777d40deb61c3214d21b32548 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Tue, 27 Oct 2026 23:31:20 +0530 Subject: [PATCH 04/12] perf: bypass event monitor hooks when disabled --- .../event/processor/event_processor.py | 18 +++++-- .../event/processor/test_blocking.py | 54 ++++++++++++------- 2 files changed, 48 insertions(+), 24 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py index ca7ede0dc5c..bda1fc51515 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py @@ -762,9 +762,20 @@ async def _process_event_queue_entry( # Set up the event context for this task. EventContext.set(entry.ctx) monitor = self._blocking_monitor + if monitor is None: + if not otel.enabled: + await self._execute_event( + entry=entry, registered_handler=registered_handler + ) + return + with otel.event_span(entry.event, entry.ctx, registered_handler): + await self._execute_event( + entry=entry, registered_handler=registered_handler + ) + return + handler = registered_handler.handler.fn - if monitor is not None: - monitor.handler_started(handler) + monitor.handler_started(handler) try: if not otel.enabled: await self._execute_event( @@ -776,8 +787,7 @@ async def _process_event_queue_entry( entry=entry, registered_handler=registered_handler ) finally: - if monitor is not None: - monitor.handler_finished(handler) + monitor.handler_finished(handler) def _create_event_task( self, diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index 7cd9280a432..a46ad03c879 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -1,10 +1,10 @@ """Tests for development-time event-loop blocking detection.""" -from collections.abc import Callable import asyncio import logging import threading import time +from collections.abc import Callable from typing import Any import pytest @@ -12,19 +12,40 @@ from reflex_base.event.processor.blocking import EventLoopBlockingMonitor -async def test_sync_handler_blocking_is_reported(caplog: pytest.LogCaptureFixture): +def _track_queued_probe( + loop: asyncio.AbstractEventLoop, monkeypatch: pytest.MonkeyPatch +) -> threading.Event: + """Return an event set after the monitor queues a responsiveness probe.""" + probe_queued = threading.Event() + call_soon_threadsafe = loop.call_soon_threadsafe + + def track_probe( + callback: Callable[..., Any], *args: Any, **kwargs: Any + ) -> asyncio.Handle: + handle = call_soon_threadsafe(callback, *args, **kwargs) + probe_queued.set() + return handle + + monkeypatch.setattr(loop, "call_soon_threadsafe", track_probe) + return probe_queued + + +async def test_sync_handler_blocking_is_reported( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): """Report the qualified name and duration of a blocking sync handler.""" + loop = asyncio.get_running_loop() + probe_queued = _track_queued_probe(loop, monkeypatch) def blocking_handler(): + assert probe_queued.wait(timeout=1) time.sleep(0.12) - monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) + monitor = EventLoopBlockingMonitor(loop, threshold=0.04) caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") monitor.handler_started(blocking_handler) monitor.start() try: - # Let the monitor queue a responsiveness probe before blocking the loop. - await asyncio.sleep(0.02) blocking_handler() await asyncio.sleep(0) finally: @@ -42,17 +63,7 @@ async def test_handler_blocking_just_over_threshold_is_detected( ): """Account for the time between responsiveness probes.""" loop = asyncio.get_running_loop() - probe_queued = threading.Event() - call_soon_threadsafe = loop.call_soon_threadsafe - - def track_probe( - callback: Callable[..., Any], *args: Any, **kwargs: Any - ) -> asyncio.Handle: - handle = call_soon_threadsafe(callback, *args, **kwargs) - probe_queued.set() - return handle - - monkeypatch.setattr(loop, "call_soon_threadsafe", track_probe) + probe_queued = _track_queued_probe(loop, monkeypatch) def blocking_handler(): assert probe_queued.wait(timeout=1) @@ -72,19 +83,22 @@ def blocking_handler(): assert "blocking_handler" in caplog.text -async def test_async_handler_blocking_is_reported(caplog: pytest.LogCaptureFixture): +async def test_async_handler_blocking_is_reported( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): """Report synchronous blocking code inside an async handler.""" + loop = asyncio.get_running_loop() + probe_queued = _track_queued_probe(loop, monkeypatch) async def blocking_handler(): + assert probe_queued.wait(timeout=1) time.sleep(0.12) - monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) + monitor = EventLoopBlockingMonitor(loop, threshold=0.04) caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") monitor.handler_started(blocking_handler) monitor.start() try: - # Let the monitor queue a responsiveness probe before blocking the loop. - await asyncio.sleep(0.02) await blocking_handler() await asyncio.sleep(0) finally: From 746ba2ae1e66ed0eed3f8b8f2b143a6dd2166cf1 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 28 Oct 2026 16:30:58 +0530 Subject: [PATCH 05/12] style: format event processor test --- .../units/reflex_base/event/processor/test_event_processor.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index 0ffca9b6e2c..d2706862d7e 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -651,9 +651,7 @@ async def test_processor_reports_blocking_sync_handler( """Report blocking handlers executed through the event queue.""" monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) mock_event_processor.event_loop_blocking_threshold = 0.04 - caplog.set_level( - logging.WARNING, logger="reflex_base.event.processor.blocking" - ) + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") async with mock_event_processor as ep: await ep.enqueue(token, Event.from_event_type(blocking_sync_event())[0]) From 49425c247008d635b1f1c75a526c588d37367b74 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 28 Oct 2026 16:40:21 +0530 Subject: [PATCH 06/12] fix: address event blocking lint feedback --- .../reflex_base/event/processor/blocking.py | 31 ++++++++++++------- .../event/processor/test_blocking.py | 4 +-- tests/units/test_app.py | 2 +- tests/units/test_config.py | 3 +- 4 files changed, 24 insertions(+), 16 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index b30a50a2b99..4c5b0feabb0 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -79,7 +79,11 @@ def handler_finished(self, handler: object) -> None: self._handlers[code] = name, count - 1 def _find_handler(self) -> str | None: - """Find the innermost active handler frame on the event-loop thread.""" + """Find the innermost active handler frame on the event-loop thread. + + Returns: + The handler name, or None if no registered handler is active. + """ frame = sys._current_frames().get(self._loop_thread_id) if frame is None: return None @@ -108,14 +112,17 @@ def _run(self) -> None: if self._stop.is_set(): return duration = time.perf_counter() - started - if duration + interval >= self.threshold and not reported: - if handler_name := self._find_handler(): - logger.warning( - "Event handler %s blocked the event loop " - "(observed delay: %.3fs; threshold: %.3fs). Move blocking " - "I/O to a thread pool or use a non-blocking async API.", - handler_name, - duration, - self.threshold, - ) - reported = True + if ( + duration + interval >= self.threshold + and not reported + and (handler_name := self._find_handler()) + ): + logger.warning( + "Event handler %s blocked the event loop " + "(observed delay: %.3fs; threshold: %.3fs). Move blocking " + "I/O to a thread pool or use a non-blocking async API.", + handler_name, + duration, + self.threshold, + ) + reported = True diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index a46ad03c879..7045f50e2ce 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -8,7 +8,6 @@ from typing import Any import pytest - from reflex_base.event.processor.blocking import EventLoopBlockingMonitor @@ -91,8 +90,9 @@ async def test_async_handler_blocking_is_reported( probe_queued = _track_queued_probe(loop, monkeypatch) async def blocking_handler(): + await asyncio.sleep(0) assert probe_queued.wait(timeout=1) - time.sleep(0.12) + time.sleep(0.12) # noqa: ASYNC251 monitor = EventLoopBlockingMonitor(loop, threshold=0.04) caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") diff --git a/tests/units/test_app.py b/tests/units/test_app.py index 61b60a09e45..20ce7ff3d9f 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -3530,7 +3530,7 @@ async def test_setup_event_processor_passes_blocking_threshold( monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.PROD.value) async with app._setup_event_processor(): - assert app.event_processor.event_loop_blocking_threshold == 0.35 + assert app.event_processor.event_loop_blocking_threshold == pytest.approx(0.35) def test_raise_on_state(): diff --git a/tests/units/test_config.py b/tests/units/test_config.py index 2e42c2ef430..72387b4d7c4 100644 --- a/tests/units/test_config.py +++ b/tests/units/test_config.py @@ -76,7 +76,8 @@ def test_event_loop_blocking_threshold_defaults_to_development_threshold( base_config_values, ): """Enable useful blocking diagnostics by default in development.""" - assert rx.Config(**base_config_values).event_loop_blocking_threshold == 0.2 + config = rx.Config(**base_config_values) + assert config.event_loop_blocking_threshold == pytest.approx(0.2) @pytest.mark.parametrize( From 97f06da4db962e77489c3d59508c216609764395 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 20:06:00 +0500 Subject: [PATCH 07/12] fix: report full blocking duration and attribute by event task Name the event task running on the blocked loop instead of tracking handler frames per event, so blocking in computed vars after the handler is reported. Log after the loop responds with the real duration. Remove the per-event hooks and the duplicated otel branch, and simplify the threshold validation. --- docs/advanced_onboarding/configuration.md | 2 +- .../reflex-base/src/reflex_base/config.py | 20 +-- .../reflex_base/event/processor/blocking.py | 105 +++++------ .../event/processor/event_processor.py | 47 ++--- .../event/processor/test_blocking.py | 165 ++++++++++-------- .../event/processor/test_event_processor.py | 29 +++ tests/units/test_config.py | 4 +- 7 files changed, 191 insertions(+), 181 deletions(-) diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index a11fa37e3ee..a11a3d44fe4 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/docs/advanced_onboarding/configuration.md @@ -27,7 +27,7 @@ See the [config reference](/docs/api-reference/config/) for all the parameters a ## 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 names the handler and 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 disabled in production. Set the threshold to `None` to disable it in development: +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 names the handler and the blocking 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 disabled in production. Set the threshold to `None` to disable it in development: ```python config = rx.Config( diff --git a/packages/reflex-base/src/reflex_base/config.py b/packages/reflex-base/src/reflex_base/config.py index 2bc18e10a4e..0c6f806aff1 100644 --- a/packages/reflex-base/src/reflex_base/config.py +++ b/packages/reflex-base/src/reflex_base/config.py @@ -403,18 +403,14 @@ def _post_init(self, **kwargs): msg = "trusted_proxy_hops must be a non-negative integer." raise ConfigError(msg) - if self.event_loop_blocking_threshold is not None: - threshold = self.event_loop_blocking_threshold - is_number = not isinstance(threshold, bool) and isinstance( - threshold, (int, float) - ) - try: - is_finite = is_number and math.isfinite(threshold) - except OverflowError: - is_finite = False - if not is_finite or threshold <= 0: - msg = "event_loop_blocking_threshold must be a positive finite number or None." - raise ConfigError(msg) + threshold = self.event_loop_blocking_threshold + if threshold is not None and ( + isinstance(threshold, bool) + or not isinstance(threshold, (int, float)) + or not 0 < threshold < math.inf + ): + msg = "event_loop_blocking_threshold must be a positive finite number or None." + raise ConfigError(msg) # Normalize route prefixes to ensure they start with a slash. self._normalize_paths() diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index 4c5b0feabb0..113073f7792 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -4,30 +4,40 @@ import asyncio import logging -import sys import threading import time -import types +from collections.abc import Mapping +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|" + class EventLoopBlockingMonitor: """Report event handlers that keep the event loop from responding.""" - def __init__(self, loop: asyncio.AbstractEventLoop, threshold: float) -> None: + 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._loop_thread_id = threading.get_ident() + self.event_handlers = event_handlers self._stop = threading.Event() - self._handlers_lock = threading.Lock() - self._handlers: dict[types.CodeType, tuple[str, int]] = {} self._thread: threading.Thread | None = None def start(self) -> None: @@ -46,54 +56,19 @@ def stop(self) -> None: self._thread.join(timeout=1) self._thread = None - def handler_started(self, handler: object) -> None: - """Record a handler while its event task is executing. - - Args: - handler: The registered handler function. - """ - code = getattr(handler, "__code__", None) - name = getattr(handler, "__qualname__", None) - if not isinstance(code, types.CodeType) or not isinstance(name, str): - return - with self._handlers_lock: - current_name, count = self._handlers.get(code, (name, 0)) - self._handlers[code] = current_name, count + 1 - - def handler_finished(self, handler: object) -> None: - """Remove a handler after its event task finishes. - - Args: - handler: The registered handler function. - """ - code = getattr(handler, "__code__", None) - if not isinstance(code, types.CodeType): - return - with self._handlers_lock: - if (current := self._handlers.get(code)) is None: - return - name, count = current - if count == 1: - del self._handlers[code] - else: - self._handlers[code] = name, count - 1 - - def _find_handler(self) -> str | None: - """Find the innermost active handler frame on the event-loop thread. + def _blocking_handler(self) -> str | None: + """Name the handler of the event task that runs on the blocked loop. Returns: - The handler name, or None if no registered handler is active. + The handler name, or None if the running task processes no event. """ - frame = sys._current_frames().get(self._loop_thread_id) - if frame is None: + task = asyncio.current_task(self.loop) + if task is None or not (name := task.get_name()).startswith(EVENT_TASK_PREFIX): return None - with self._handlers_lock: - handlers = self._handlers.copy() - while frame is not None: - if handler := handlers.get(frame.f_code): - return handler[0] - frame = frame.f_back - return None + event_name = name[len(EVENT_TASK_PREFIX) :].partition("|")[0] + if (registered := self.event_handlers.get(event_name)) is None: + return event_name + return registered.handler.fn.__qualname__ def _run(self) -> None: """Poll loop responsiveness from outside the event-loop thread.""" @@ -107,22 +82,24 @@ def _run(self) -> None: except RuntimeError: return - reported = False + # The task is only visible while the loop is blocked, so name it + # then and report the full duration once the loop responds. + handler_name = None while not responsive.wait(interval): if self._stop.is_set(): return - duration = time.perf_counter() - started if ( - duration + interval >= self.threshold - and not reported - and (handler_name := self._find_handler()) + handler_name is None + and time.perf_counter() - started + interval >= self.threshold ): - logger.warning( - "Event handler %s blocked the event loop " - "(observed delay: %.3fs; threshold: %.3fs). Move blocking " - "I/O to a thread pool or use a non-blocking async API.", - handler_name, - duration, - self.threshold, - ) - reported = True + handler_name = self._blocking_handler() + if handler_name is not None: + logger.warning( + "Event handler %s blocked the event loop for %.3fs " + "(threshold: %.3fs), including state updates such as computed " + "vars. Move blocking I/O to a thread pool or use a " + "non-blocking async API.", + handler_name, + time.perf_counter() - started, + self.threshold, + ) diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py index bda1fc51515..7124d00d6b6 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py @@ -20,7 +20,10 @@ 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 EventLoopBlockingMonitor +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 @@ -254,7 +257,9 @@ async def start(self) -> None: and environment.REFLEX_ENV_MODE.get() == constants.Env.DEV ): self._blocking_monitor = EventLoopBlockingMonitor( - asyncio.get_running_loop(), self.event_loop_blocking_threshold + asyncio.get_running_loop(), + self.event_loop_blocking_threshold, + RegistrationContext.get().event_handlers, ) self._blocking_monitor.start() self._ensure_queue_task() @@ -761,33 +766,15 @@ async def _process_event_queue_entry( """ # Set up the event context for this task. EventContext.set(entry.ctx) - monitor = self._blocking_monitor - if monitor is None: - if not otel.enabled: - await self._execute_event( - entry=entry, registered_handler=registered_handler - ) - return - with otel.event_span(entry.event, entry.ctx, registered_handler): - await self._execute_event( - entry=entry, registered_handler=registered_handler - ) + if not otel.enabled: + await self._execute_event( + entry=entry, registered_handler=registered_handler + ) return - - handler = registered_handler.handler.fn - monitor.handler_started(handler) - try: - if not otel.enabled: - await self._execute_event( - entry=entry, registered_handler=registered_handler - ) - return - with otel.event_span(entry.event, entry.ctx, registered_handler): - await self._execute_event( - entry=entry, registered_handler=registered_handler - ) - finally: - monitor.handler_finished(handler) + with otel.event_span(entry.event, entry.ctx, registered_handler): + await self._execute_event( + entry=entry, registered_handler=registered_handler + ) def _create_event_task( self, @@ -807,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) diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index 7045f50e2ce..dabf8eeca9b 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -4,11 +4,17 @@ import logging import threading import time -from collections.abc import Callable +from collections.abc import Callable, Coroutine from typing import Any import pytest -from reflex_base.event.processor.blocking import EventLoopBlockingMonitor +from reflex_base.event.processor.blocking import ( + EVENT_TASK_PREFIX, + EventLoopBlockingMonitor, +) +from reflex_base.registry import RegisteredEventHandler + +from reflex.event import EventHandler def _track_queued_probe( @@ -29,101 +35,116 @@ def track_probe( return probe_queued -async def test_sync_handler_blocking_is_reported( - caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch -): - """Report the qualified name and duration of a blocking sync handler.""" - loop = asyncio.get_running_loop() - probe_queued = _track_queued_probe(loop, monkeypatch) - - def blocking_handler(): - assert probe_queued.wait(timeout=1) - time.sleep(0.12) - - monitor = EventLoopBlockingMonitor(loop, threshold=0.04) - caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") - monitor.handler_started(blocking_handler) +async def _run_monitored( + fn: Callable[[], Coroutine[Any, Any, None]], + task_name: str, + event_handlers: dict[str, RegisteredEventHandler] | None = None, +) -> None: + """Run fn in a task with the given name while a monitor watches the loop.""" + monitor = EventLoopBlockingMonitor( + asyncio.get_running_loop(), threshold=0.04, event_handlers=event_handlers or {} + ) monitor.start() try: - blocking_handler() + await asyncio.create_task(fn(), name=task_name) await asyncio.sleep(0) finally: - monitor.handler_finished(blocking_handler) monitor.stop() - assert "blocking_handler" in caplog.text - assert "blocked the event loop" in caplog.text - assert "thread pool or use a non-blocking async API" in caplog.text +def _blocking( + probe_queued: threading.Event, duration: float +) -> Callable[[], Coroutine[Any, Any, None]]: + """Create a coroutine function that blocks the loop after a probe is queued. -async def test_handler_blocking_just_over_threshold_is_detected( - caplog: pytest.LogCaptureFixture, - monkeypatch: pytest.MonkeyPatch, -): - """Account for the time between responsiveness probes.""" - loop = asyncio.get_running_loop() - probe_queued = _track_queued_probe(loop, monkeypatch) + Returns: + The blocking coroutine function. + """ - def blocking_handler(): + async def blocking_handler(): # noqa: RUF029 assert probe_queued.wait(timeout=1) - time.sleep(0.055) + time.sleep(duration) # noqa: ASYNC251 + + return blocking_handler + + +@pytest.fixture +def caplog_blocking(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture: + """Capture warnings of the blocking monitor. - monitor = EventLoopBlockingMonitor(loop, threshold=0.04) + Returns: + The log capture fixture. + """ caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") - monitor.handler_started(blocking_handler) - monitor.start() - try: - blocking_handler() - await asyncio.sleep(0) - finally: - monitor.handler_finished(blocking_handler) - monitor.stop() + return caplog - assert "blocking_handler" in caplog.text +async def test_blocking_event_is_reported_with_handler_name( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): + """Name the registered handler of the blocking event task.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) + handler = _blocking(probe_queued, 0.12) + registered = RegisteredEventHandler(handler=EventHandler(fn=handler), states=()) -async def test_async_handler_blocking_is_reported( - caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch + await _run_monitored( + handler, + f"{EVENT_TASK_PREFIX}state.blocking|token|txid", + {"state.blocking": registered}, + ) + + assert "blocking_handler" in caplog_blocking.text + assert "blocked the event loop for 0.1" in caplog_blocking.text + assert "thread pool or use a non-blocking async API" in caplog_blocking.text + + +async def test_unregistered_event_is_reported_with_event_name( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch ): - """Report synchronous blocking code inside an async handler.""" - loop = asyncio.get_running_loop() - probe_queued = _track_queued_probe(loop, monkeypatch) + """Fall back to the event name when no handler is registered for it.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) - async def blocking_handler(): - await asyncio.sleep(0) - assert probe_queued.wait(timeout=1) - time.sleep(0.12) # noqa: ASYNC251 + await _run_monitored( + _blocking(probe_queued, 0.12), f"{EVENT_TASK_PREFIX}state.unknown|token|txid" + ) - monitor = EventLoopBlockingMonitor(loop, threshold=0.04) - caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") - monitor.handler_started(blocking_handler) - monitor.start() - try: - await blocking_handler() - await asyncio.sleep(0) - finally: - monitor.handler_finished(blocking_handler) - monitor.stop() + assert "Event handler state.unknown blocked" in caplog_blocking.text + + +async def test_blocking_just_over_threshold_is_detected( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): + """Account for the time between responsiveness probes.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) + + await _run_monitored( + _blocking(probe_queued, 0.055), f"{EVENT_TASK_PREFIX}state.blocking|token|txid" + ) - assert "blocking_handler" in caplog.text + assert "state.blocking" in caplog_blocking.text -async def test_async_handler_await_does_not_report_as_blocking( - caplog: pytest.LogCaptureFixture, +async def test_blocking_outside_event_task_is_not_reported( + caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +): + """Only report tasks that process an event.""" + probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) + + await _run_monitored(_blocking(probe_queued, 0.12), "other_task") + + assert "blocked the event loop" not in caplog_blocking.text + + +async def test_awaiting_event_is_not_reported( + caplog_blocking: pytest.LogCaptureFixture, ): """Do not report handlers that yield control while doing async work.""" async def yielding_handler(): await asyncio.sleep(0.1) - monitor = EventLoopBlockingMonitor(asyncio.get_running_loop(), threshold=0.04) - caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") - monitor.handler_started(yielding_handler) - monitor.start() - try: - await yielding_handler() - finally: - monitor.handler_finished(yielding_handler) - monitor.stop() + await _run_monitored( + yielding_handler, f"{EVENT_TASK_PREFIX}state.yielding|token|txid" + ) - assert "blocked the event loop" not in caplog.text + assert "blocked the event loop" not in caplog_blocking.text diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index d2706862d7e..3d4ba5028bf 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -4,6 +4,7 @@ import contextlib import dataclasses import logging +import re import sys import time from collections.abc import AsyncIterator, Mapping @@ -658,6 +659,34 @@ async def test_processor_reports_blocking_sync_handler( assert "_blocking_sync_handler" in caplog.text assert "blocked the event loop" in caplog.text + duration = re.search(r"blocked the event loop for (\d+\.\d+)s", caplog.text) + assert duration is not None + assert float(duration.group(1)) >= 0.1 + + +async def test_processor_reports_blocking_after_handler_returns( + mock_event_processor: EventProcessor, + token: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +): + """Attribute blocking in post-handler processing (e.g. computed vars) to the event.""" + execute_event = EventProcessor._execute_event + + async def _execute_event(self: EventProcessor, **kwargs): + await execute_event(self, **kwargs) + time.sleep(0.12) # noqa: ASYNC251 + + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + monkeypatch.setattr(EventProcessor, "_execute_event", _execute_event) + mock_event_processor.event_loop_blocking_threshold = 0.04 + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + + async with mock_event_processor as ep: + await ep.enqueue(token, Event.from_event_type(noop_event())[0]) + + assert "_noop_handler" in caplog.text + assert "blocked the event loop" in caplog.text async def test_enqueue_returns_future( diff --git a/tests/units/test_config.py b/tests/units/test_config.py index 72387b4d7c4..5f953f55f8d 100644 --- a/tests/units/test_config.py +++ b/tests/units/test_config.py @@ -80,9 +80,7 @@ def test_event_loop_blocking_threshold_defaults_to_development_threshold( assert config.event_loop_blocking_threshold == pytest.approx(0.2) -@pytest.mark.parametrize( - "threshold", [0, -1, float("inf"), float("nan"), True, "1", 10**1000] -) +@pytest.mark.parametrize("threshold", [0, -1, float("inf"), float("nan"), True, "1"]) def test_event_loop_blocking_threshold_must_be_positive_finite_or_none( base_config_values, threshold ): From 191cd00d0127d1045072a821fe0ee046f7ec0f1c Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 21:39:16 +0500 Subject: [PATCH 08/12] feat: show the blocking stack in event loop blocking warnings Capture the loop thread's frames from the event task's entry to the blocking line when the threshold is crossed, so the warning points at the code that blocked, including computed vars and C-extension calls. --- docs/advanced_onboarding/configuration.md | 2 +- .../reflex_base/event/processor/blocking.py | 57 ++++++++++++++----- .../event/processor/test_blocking.py | 2 + .../event/processor/test_event_processor.py | 5 ++ 4 files changed, 52 insertions(+), 14 deletions(-) diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index a11a3d44fe4..4faacabb2a4 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/docs/advanced_onboarding/configuration.md @@ -27,7 +27,7 @@ See the [config reference](/docs/api-reference/config/) for all the parameters a ## 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 names the handler and the blocking 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 disabled in production. Set the threshold to `None` to disable it in development: +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 names the handler, the blocking duration, and the stack of the code that blocked. The duration 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 disabled in production. Set the threshold to `None` to disable it in development: ```python config = rx.Config( diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index 113073f7792..b6299d3fe2e 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -4,8 +4,10 @@ import asyncio import logging +import sys import threading import time +import traceback from collections.abc import Mapping from typing import TYPE_CHECKING @@ -37,6 +39,8 @@ def __init__( self.loop = loop self.threshold = threshold self.event_handlers = event_handlers + # The monitor is created on the loop thread. + self._loop_thread_id = threading.get_ident() self._stop = threading.Event() self._thread: threading.Thread | None = None @@ -56,19 +60,43 @@ def stop(self) -> None: self._thread.join(timeout=1) self._thread = None - def _blocking_handler(self) -> str | None: - """Name the handler of the event task that runs on the blocked loop. + def _describe_blocking(self) -> tuple[str, str] | None: + """Describe the event task that runs on the blocked loop. Returns: - The handler name, or None if the running task processes no event. + The handler name and the formatted stack of the event task, or None + if the running task processes no event. """ task = asyncio.current_task(self.loop) if task is None or not (name := task.get_name()).startswith(EVENT_TASK_PREFIX): return None event_name = name[len(EVENT_TASK_PREFIX) :].partition("|")[0] - if (registered := self.event_handlers.get(event_name)) is None: - return event_name - return registered.handler.fn.__qualname__ + registered = self.event_handlers.get(event_name) + handler_name = ( + event_name if registered is None else registered.handler.fn.__qualname__ + ) + return handler_name, "".join(traceback.format_list(self._task_stack(task))) + + def _task_stack(self, task: asyncio.Task) -> traceback.StackSummary: + """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. + """ + 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: + break + frame = frame.f_back + return traceback.StackSummary.extract( + (frame, frame.f_lineno) for frame in reversed(frames) + ) def _run(self) -> None: """Poll loop responsiveness from outside the event-loop thread.""" @@ -82,24 +110,27 @@ def _run(self) -> None: except RuntimeError: return - # The task is only visible while the loop is blocked, so name it - # then and report the full duration once the loop responds. - handler_name = None + # The task is only visible while the loop is blocked, so describe + # it then and report the full duration once the loop responds. + blocking = None while not responsive.wait(interval): if self._stop.is_set(): return if ( - handler_name is None + blocking is None and time.perf_counter() - started + interval >= self.threshold ): - handler_name = self._blocking_handler() - if handler_name is not None: + blocking = self._describe_blocking() + if blocking is not None: + handler_name, stack = blocking logger.warning( "Event handler %s blocked the event loop for %.3fs " "(threshold: %.3fs), including state updates such as computed " "vars. Move blocking I/O to a thread pool or use a " - "non-blocking async API.", + "non-blocking async API.\n" + "Blocking stack (most recent call last):\n%s", handler_name, time.perf_counter() - started, self.threshold, + stack.rstrip(), ) diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index dabf8eeca9b..08551ece40c 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -95,6 +95,8 @@ async def test_blocking_event_is_reported_with_handler_name( assert "blocking_handler" in caplog_blocking.text assert "blocked the event loop for 0.1" in caplog_blocking.text + assert "time.sleep(duration) # noqa: ASYNC251" in caplog_blocking.text + assert "_run_monitored" not in caplog_blocking.text assert "thread pool or use a non-blocking async API" in caplog_blocking.text diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index 3d4ba5028bf..4b0090c01cd 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -662,6 +662,10 @@ async def test_processor_reports_blocking_sync_handler( duration = re.search(r"blocked the event loop for (\d+\.\d+)s", caplog.text) assert duration is not None assert float(duration.group(1)) >= 0.1 + # The stack of the event task ends at the blocking line, without loop frames. + assert "Blocking stack (most recent call last):" in caplog.text + assert "time.sleep(0.12)" in caplog.text + assert "base_events.py" not in caplog.text async def test_processor_reports_blocking_after_handler_returns( @@ -687,6 +691,7 @@ async def _execute_event(self: EventProcessor, **kwargs): assert "_noop_handler" in caplog.text assert "blocked the event loop" in caplog.text + assert "time.sleep(0.12) # noqa: ASYNC251" in caplog.text async def test_enqueue_returns_future( From 3f51a4c3e394896134b3bc61834ef50846c918d6 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 21:55:09 +0500 Subject: [PATCH 09/12] fix: warn while the loop is still blocked and drop stale stacks Log the handler and its stack as soon as the threshold is reached, so a handler that never returns is reported, then log the total duration once the loop responds. Report nothing when the frame snapshot no longer contains the event task's entry frame. --- docs/advanced_onboarding/configuration.md | 2 +- .../reflex_base/event/processor/blocking.py | 54 ++++++++++-------- .../event/processor/test_blocking.py | 56 ++++++++++++++++++- 3 files changed, 86 insertions(+), 26 deletions(-) diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index 4faacabb2a4..e8d53743505 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/docs/advanced_onboarding/configuration.md @@ -27,7 +27,7 @@ See the [config reference](/docs/api-reference/config/) for all the parameters a ## 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 names the handler, the blocking duration, and the stack of the code that blocked. The duration 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 disabled in production. Set the threshold to `None` to disable it in development: +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 shows the stack of the code that blocked. 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 disabled in production. Set the threshold to `None` to disable it in development: ```python config = rx.Config( diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index b6299d3fe2e..2977964833e 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -65,26 +65,29 @@ def _describe_blocking(self) -> tuple[str, str] | None: Returns: The handler name and the formatted stack of the event task, or None - if the running task processes no event. + if the running task processes no event or has left the loop. """ task = asyncio.current_task(self.loop) if task is None or not (name := task.get_name()).startswith(EVENT_TASK_PREFIX): return None event_name = name[len(EVENT_TASK_PREFIX) :].partition("|")[0] registered = self.event_handlers.get(event_name) + if (stack := self._task_stack(task)) is None: + return None handler_name = ( event_name if registered is None else registered.handler.fn.__qualname__ ) - return handler_name, "".join(traceback.format_list(self._task_stack(task))) + return handler_name, "".join(traceback.format_list(stack)) - def _task_stack(self, task: asyncio.Task) -> traceback.StackSummary: + 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. + 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) @@ -92,11 +95,11 @@ def _task_stack(self, task: asyncio.Task) -> traceback.StackSummary: while frame is not None: frames.append(frame) if frame is entry: - break + return traceback.StackSummary.extract( + (frame, frame.f_lineno) for frame in reversed(frames) + ) frame = frame.f_back - return traceback.StackSummary.extract( - (frame, frame.f_lineno) for frame in reversed(frames) - ) + return None def _run(self) -> None: """Poll loop responsiveness from outside the event-loop thread.""" @@ -110,27 +113,32 @@ def _run(self) -> None: except RuntimeError: return - # The task is only visible while the loop is blocked, so describe - # it then and report the full duration once the loop responds. - blocking = None + # The task is only visible while the loop is blocked, so report it + # then, and report the total duration once the loop responds. + handler_name = None while not responsive.wait(interval): if self._stop.is_set(): return + elapsed = time.perf_counter() - started if ( - blocking is None - and time.perf_counter() - started + interval >= self.threshold + handler_name is None + and elapsed + interval >= self.threshold + and (blocking := self._describe_blocking()) is not None ): - blocking = self._describe_blocking() - if blocking is not None: - handler_name, stack = blocking + handler_name, stack = blocking + logger.warning( + "Event handler %s is blocking the event loop (threshold: " + "%.3fs), including state updates such as computed vars. " + "Move blocking I/O to a thread pool or use a non-blocking " + "async API.\n" + "Blocking stack (most recent call last):\n%s", + handler_name, + self.threshold, + stack.rstrip(), + ) + if handler_name is not None: logger.warning( - "Event handler %s blocked the event loop for %.3fs " - "(threshold: %.3fs), including state updates such as computed " - "vars. Move blocking I/O to a thread pool or use a " - "non-blocking async API.\n" - "Blocking stack (most recent call last):\n%s", + "Event handler %s blocked the event loop for %.3fs in total.", handler_name, time.perf_counter() - started, - self.threshold, - stack.rstrip(), ) diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index 08551ece40c..68f33b6e57c 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -2,6 +2,7 @@ import asyncio import logging +import sys import threading import time from collections.abc import Callable, Coroutine @@ -134,7 +135,7 @@ async def test_blocking_outside_event_task_is_not_reported( await _run_monitored(_blocking(probe_queued, 0.12), "other_task") - assert "blocked the event loop" not in caplog_blocking.text + assert "the event loop" not in caplog_blocking.text async def test_awaiting_event_is_not_reported( @@ -149,4 +150,55 @@ async def yielding_handler(): yielding_handler, f"{EVENT_TASK_PREFIX}state.yielding|token|txid" ) - assert "blocked the event loop" not in caplog_blocking.text + assert "the event loop" not in caplog_blocking.text + + +async def test_hung_event_is_reported_while_still_blocked( + caplog_blocking: pytest.LogCaptureFixture, +): + """Warn when the threshold is reached, before the loop responds again.""" + warned = threading.Event() + + class _SetOnWarning(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + warned.set() + + handler = _SetOnWarning(level=logging.WARNING) + blocking_logger = logging.getLogger("reflex_base.event.processor.blocking") + blocking_logger.addHandler(handler) + + async def hung_handler(): # noqa: RUF029 + # Stands in for a handler that never releases the loop. + assert warned.wait(timeout=1) + + try: + await _run_monitored(hung_handler, f"{EVENT_TASK_PREFIX}state.hung|token|txid") + finally: + blocking_logger.removeHandler(handler) + + assert "state.hung is blocking the event loop" in caplog_blocking.text + assert "warned.wait(timeout=1)" in caplog_blocking.text + + +async def test_stack_without_task_entry_is_not_reported( + monkeypatch: pytest.MonkeyPatch, +): + """Do not describe a frame snapshot taken after the task left the loop.""" + monitor = EventLoopBlockingMonitor( + asyncio.get_running_loop(), threshold=0.04, event_handlers={} + ) + unrelated = sys._getframe() + monkeypatch.setattr( + sys, + "_current_frames", + lambda: {monitor._loop_thread_id: unrelated}, + ) + + async def handler(): # noqa: RUF029 + return monitor._describe_blocking() + + description = await asyncio.create_task( + handler(), name=f"{EVENT_TASK_PREFIX}state.moved|token|txid" + ) + + assert description is None From b0e46a139cd28a175bcd89f21441e82fb8bfedc1 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 22:13:32 +0500 Subject: [PATCH 10/12] test: document nested helpers in blocking monitor tests --- .../event/processor/test_blocking.py | 24 ++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index 68f33b6e57c..00b7229b050 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -28,6 +28,16 @@ def _track_queued_probe( def track_probe( callback: Callable[..., Any], *args: Any, **kwargs: Any ) -> asyncio.Handle: + """Queue the callback and record that a probe was queued. + + Args: + callback: The callback to queue. + *args: Positional arguments for the callback. + **kwargs: Keyword arguments for call_soon_threadsafe. + + Returns: + The handle of the queued callback. + """ handle = call_soon_threadsafe(callback, *args, **kwargs) probe_queued.set() return handle @@ -63,6 +73,7 @@ def _blocking( """ async def blocking_handler(): # noqa: RUF029 + """Block the loop for the duration after a probe is queued.""" assert probe_queued.wait(timeout=1) time.sleep(duration) # noqa: ASYNC251 @@ -144,6 +155,7 @@ async def test_awaiting_event_is_not_reported( """Do not report handlers that yield control while doing async work.""" async def yielding_handler(): + """Yield to the loop while waiting.""" await asyncio.sleep(0.1) await _run_monitored( @@ -161,6 +173,11 @@ async def test_hung_event_is_reported_while_still_blocked( class _SetOnWarning(logging.Handler): def emit(self, record: logging.LogRecord) -> None: + """Record that the monitor logged a warning. + + Args: + record: The log record. + """ warned.set() handler = _SetOnWarning(level=logging.WARNING) @@ -168,7 +185,7 @@ def emit(self, record: logging.LogRecord) -> None: blocking_logger.addHandler(handler) async def hung_handler(): # noqa: RUF029 - # Stands in for a handler that never releases the loop. + """Hold the loop until the monitor warns, like a handler that never returns.""" assert warned.wait(timeout=1) try: @@ -195,6 +212,11 @@ async def test_stack_without_task_entry_is_not_reported( ) async def handler(): # noqa: RUF029 + """Describe the blocking from inside the event task. + + Returns: + The description of the blocking. + """ return monitor._describe_blocking() description = await asyncio.create_task( From e192a31306fa7cbaa3961c12ddc7ee9bd246d402 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 28 Oct 2026 22:51:17 +0500 Subject: [PATCH 11/12] feat: disable blocking detection with 0 and name the app line Use 0 instead of None to turn detection off, so REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD=0 works: the env parser cannot produce None for optional fields. Name the innermost frame of app code (under the app root, outside the stdlib and installed packages) in the first warning line, so library frames do not hide the line to fix. --- docs/advanced_onboarding/configuration.md | 6 +-- .../reflex-base/src/reflex_base/config.py | 10 ++-- .../reflex_base/event/processor/blocking.py | 49 ++++++++++++++++--- .../event/processor/event_processor.py | 4 +- .../event/processor/test_event_processor.py | 35 +++++++++++++ tests/units/test_config.py | 21 +++++--- 6 files changed, 103 insertions(+), 22 deletions(-) diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index e8d53743505..64a7fe55ea6 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/docs/advanced_onboarding/configuration.md @@ -27,16 +27,16 @@ See the [config reference](/docs/api-reference/config/) for all the parameters a ## 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 shows the stack of the code that blocked. 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 disabled in production. Set the threshold to `None` to disable it in development: +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. 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=None, + event_loop_blocking_threshold=0, ) ``` -The setting can also be overridden with `REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD`. +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 diff --git a/packages/reflex-base/src/reflex_base/config.py b/packages/reflex-base/src/reflex_base/config.py index 0c6f806aff1..524fb4b3098 100644 --- a/packages/reflex-base/src/reflex_base/config.py +++ b/packages/reflex-base/src/reflex_base/config.py @@ -195,7 +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 None to disable detection. + event_loop_blocking_threshold: Minimum event-loop blocking duration in seconds to report in development. Set to 0 to disable detection. """ app_name: str @@ -287,7 +287,7 @@ class BaseConfig: transport: Literal["websocket", "polling"] = "websocket" - event_loop_blocking_threshold: float | None = 0.2 + event_loop_blocking_threshold: float = 0.2 # Whether to skip plugin checks. _skip_plugins_checks: bool = dataclasses.field(default=False, repr=False) @@ -404,12 +404,12 @@ def _post_init(self, **kwargs): raise ConfigError(msg) threshold = self.event_loop_blocking_threshold - if threshold is not None and ( + if ( isinstance(threshold, bool) or not isinstance(threshold, (int, float)) - or not 0 < threshold < math.inf + or not 0 <= threshold < math.inf ): - msg = "event_loop_blocking_threshold must be a positive finite number or None." + 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. diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index 2977964833e..0222d369d8f 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -4,11 +4,14 @@ 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: @@ -19,6 +22,12 @@ # 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 event handlers that keep the event loop from responding.""" @@ -39,8 +48,9 @@ def __init__( self.loop = loop self.threshold = threshold self.event_handlers = event_handlers - # The monitor is created on the loop thread. + # 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 @@ -60,11 +70,12 @@ def stop(self) -> None: self._thread.join(timeout=1) self._thread = None - def _describe_blocking(self) -> tuple[str, str] | None: + def _describe_blocking(self) -> tuple[str, str, str] | None: """Describe the event task that runs on the blocked loop. Returns: - The handler name and the formatted stack of the event task, or None + The handler name, the location of the innermost app frame (or an + empty string), and the formatted stack of the event task, or None if the running task processes no event or has left the loop. """ task = asyncio.current_task(self.loop) @@ -77,7 +88,32 @@ def _describe_blocking(self) -> tuple[str, str] | None: handler_name = ( event_name if registered is None else registered.handler.fn.__qualname__ ) - return handler_name, "".join(traceback.format_list(stack)) + 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 handler_name, 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. @@ -125,14 +161,15 @@ def _run(self) -> None: and elapsed + interval >= self.threshold and (blocking := self._describe_blocking()) is not None ): - handler_name, stack = blocking + handler_name, location, stack = blocking logger.warning( - "Event handler %s is blocking the event loop (threshold: " + "Event handler %s is blocking the event loop%s (threshold: " "%.3fs), including state updates such as computed vars. " "Move blocking I/O to a thread pool or use a non-blocking " "async API.\n" "Blocking stack (most recent call last):\n%s", handler_name, + location, self.threshold, stack.rstrip(), ) diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py index 7124d00d6b6..3b76861899d 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py @@ -113,7 +113,7 @@ class EventProcessor: Callable[[Exception], EventSpec | list[EventSpec] | None] | None ) = None graceful_shutdown_timeout: float | None = None - event_loop_blocking_threshold: float | None = None + event_loop_blocking_threshold: float = 0 _queue: asyncio.Queue[EventQueueEntry] | None = dataclasses.field( default=None, init=False @@ -253,7 +253,7 @@ async def start(self) -> None: self._attached_root_context_token = EventContext.set(self._root_context) self._queue = asyncio.Queue() if ( - self.event_loop_blocking_threshold is not None + self.event_loop_blocking_threshold and environment.REFLEX_ENV_MODE.get() == constants.Env.DEV ): self._blocking_monitor = EventLoopBlockingMonitor( diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index 4b0090c01cd..85045933672 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -6,6 +6,7 @@ import logging import re import sys +import threading import time from collections.abc import AsyncIterator, Mapping from typing import Any @@ -49,6 +50,11 @@ def _blocking_sync_handler(): time.sleep(0.12) +def _blocking_stdlib_handler(): + """Block the event loop inside a stdlib call for the detector test.""" + threading.Event().wait(0.12) + + async def _error_handler(): # noqa: RUF029 """A handler that always raises.""" raise RuntimeError("boom") # noqa: EM101 @@ -372,6 +378,7 @@ async def _counting_superseding_handler(tick: int = 0, limit: int = 3): noop_event = EventHandler(fn=_noop_handler) slow_event = EventHandler(fn=_slow_handler) blocking_sync_event = EventHandler(fn=_blocking_sync_handler) +blocking_stdlib_event = EventHandler(fn=_blocking_stdlib_handler) error_event = EventHandler(fn=_error_handler) recovery_event = EventHandler(fn=_recovery_handler) logging_event = EventHandler(fn=_logging_handler) @@ -413,6 +420,7 @@ def _register_handlers(forked_registration_context: RegistrationContext): noop_event, slow_event, blocking_sync_event, + blocking_stdlib_event, error_event, recovery_event, logging_event, @@ -546,6 +554,12 @@ async def test_blocking_monitor_starts_only_in_development( await dev_processor.stop() assert dev_processor._blocking_monitor is None + disabled_processor = EventProcessor(event_loop_blocking_threshold=0) + disabled_processor.configure() + await disabled_processor.start() + assert disabled_processor._blocking_monitor is None + await disabled_processor.stop() + async def test_stop_idempotent(processor: EventProcessor): """Stopping an already-stopped processor does not error. @@ -668,6 +682,27 @@ async def test_processor_reports_blocking_sync_handler( assert "base_events.py" not in caplog.text +async def test_processor_reports_innermost_app_frame( + mock_event_processor: EventProcessor, + token: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +): + """Name the app line that called the blocking library code.""" + monkeypatch.setenv("REFLEX_ENV_MODE", constants.Env.DEV.value) + mock_event_processor.event_loop_blocking_threshold = 0.04 + caplog.set_level(logging.WARNING, logger="reflex_base.event.processor.blocking") + + async with mock_event_processor as ep: + await ep.enqueue(token, Event.from_event_type(blocking_stdlib_event())[0]) + + assert ( + "is blocking the event loop at " + "tests/units/reflex_base/event/processor/test_event_processor.py:" + ) in caplog.text + assert "in _blocking_stdlib_handler (threshold" in caplog.text + + async def test_processor_reports_blocking_after_handler_returns( mock_event_processor: EventProcessor, token: str, diff --git a/tests/units/test_config.py b/tests/units/test_config.py index 5f953f55f8d..2d94059226f 100644 --- a/tests/units/test_config.py +++ b/tests/units/test_config.py @@ -80,22 +80,31 @@ def test_event_loop_blocking_threshold_defaults_to_development_threshold( assert config.event_loop_blocking_threshold == pytest.approx(0.2) -@pytest.mark.parametrize("threshold", [0, -1, float("inf"), float("nan"), True, "1"]) -def test_event_loop_blocking_threshold_must_be_positive_finite_or_none( +@pytest.mark.parametrize("threshold", [-1, float("inf"), float("nan"), True, "1", None]) +def test_event_loop_blocking_threshold_must_be_non_negative_finite( base_config_values, threshold ): """Reject invalid event-loop blocking thresholds.""" with pytest.raises( ConfigError, - match="event_loop_blocking_threshold must be a positive finite number or None", + match="event_loop_blocking_threshold must be a non-negative finite number", ): rx.Config(**base_config_values, event_loop_blocking_threshold=threshold) def test_event_loop_blocking_detection_can_be_disabled(base_config_values): - """Allow apps to disable event-loop blocking detection.""" - config = rx.Config(**base_config_values, event_loop_blocking_threshold=None) - assert config.event_loop_blocking_threshold is None + """Allow apps to disable event-loop blocking detection with 0.""" + config = rx.Config(**base_config_values, event_loop_blocking_threshold=0) + assert config.event_loop_blocking_threshold == 0 + + +def test_event_loop_blocking_detection_can_be_disabled_from_env( + base_config_values, monkeypatch: pytest.MonkeyPatch +): + """Allow the environment to disable event-loop blocking detection.""" + monkeypatch.setenv("REFLEX_EVENT_LOOP_BLOCKING_THRESHOLD", "0") + config = rx.Config(**base_config_values) + assert config.update_from_env()["event_loop_blocking_threshold"] == 0 @pytest.mark.parametrize("trusted_proxy_hops", [-1, True, False, 1.5, "2", None]) From a1e0579e0da9446c5635bb4fad04fd98bcbd887c Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 29 Oct 2026 19:48:53 +0500 Subject: [PATCH 12/12] feat: report blocking in any task, not only event tasks A blocking task that processes no event, such as the lock waiter or a task the app starts, is now reported by its task name. Event handler warnings keep their wording. --- docs/advanced_onboarding/configuration.md | 2 +- .../reflex_base/event/processor/blocking.py | 55 ++++++++++--------- .../event/processor/test_blocking.py | 10 +++- .../event/processor/test_event_processor.py | 9 +-- 4 files changed, 43 insertions(+), 33 deletions(-) diff --git a/docs/advanced_onboarding/configuration.md b/docs/advanced_onboarding/configuration.md index 64a7fe55ea6..c69d1361fa0 100644 --- a/docs/advanced_onboarding/configuration.md +++ b/docs/advanced_onboarding/configuration.md @@ -27,7 +27,7 @@ See the [config reference](/docs/api-reference/config/) for all the parameters a ## 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. 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: +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( diff --git a/packages/reflex-base/src/reflex_base/event/processor/blocking.py b/packages/reflex-base/src/reflex_base/event/processor/blocking.py index 0222d369d8f..840d3fe149b 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/blocking.py +++ b/packages/reflex-base/src/reflex_base/event/processor/blocking.py @@ -1,4 +1,4 @@ -"""Development-time detection of event handlers blocking the asyncio loop.""" +"""Development-time detection of tasks blocking the asyncio loop.""" from __future__ import annotations @@ -30,7 +30,7 @@ class EventLoopBlockingMonitor: - """Report event handlers that keep the event loop from responding.""" + """Report tasks, such as event handlers, that keep the event loop from responding.""" def __init__( self, @@ -70,24 +70,29 @@ def stop(self) -> None: self._thread.join(timeout=1) self._thread = None - def _describe_blocking(self) -> tuple[str, str, str] | None: - """Describe the event task that runs on the blocked loop. + def _describe_blocking(self) -> tuple[str, str, str, str] | None: + """Describe the task that runs on the blocked loop. Returns: - The handler name, the location of the innermost app frame (or an - empty string), and the formatted stack of the event task, or None - if the running task processes no event or has left the loop. + 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 not (name := task.get_name()).startswith(EVENT_TASK_PREFIX): + if task is None or (stack := self._task_stack(task)) is None: return None - event_name = name[len(EVENT_TASK_PREFIX) :].partition("|")[0] - registered = self.event_handlers.get(event_name) - if (stack := self._task_stack(task)) is None: - return None - handler_name = ( - event_name if registered is None else registered.handler.fn.__qualname__ - ) + 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 = ( "" @@ -95,7 +100,7 @@ def _describe_blocking(self) -> tuple[str, str, str] | None: else f" at {app_frame.filename[len(self._app_root) :]}:" f"{app_frame.lineno} in {app_frame.name}" ) - return handler_name, location, "".join(traceback.format_list(stack)) + return subject, scope, location, "".join(traceback.format_list(stack)) def _app_frame( self, stack: traceback.StackSummary @@ -151,31 +156,31 @@ def _run(self) -> None: # The task is only visible while the loop is blocked, so report it # then, and report the total duration once the loop responds. - handler_name = None + subject = None while not responsive.wait(interval): if self._stop.is_set(): return elapsed = time.perf_counter() - started if ( - handler_name is None + subject is None and elapsed + interval >= self.threshold and (blocking := self._describe_blocking()) is not None ): - handler_name, location, stack = blocking + subject, scope, location, stack = blocking logger.warning( - "Event handler %s is blocking the event loop%s (threshold: " - "%.3fs), including state updates such as computed vars. " + "%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", - handler_name, + subject, location, self.threshold, + scope, stack.rstrip(), ) - if handler_name is not None: + if subject is not None: logger.warning( - "Event handler %s blocked the event loop for %.3fs in total.", - handler_name, + "%s blocked the event loop for %.3fs in total.", + subject, time.perf_counter() - started, ) diff --git a/tests/units/reflex_base/event/processor/test_blocking.py b/tests/units/reflex_base/event/processor/test_blocking.py index 00b7229b050..084c42daf16 100644 --- a/tests/units/reflex_base/event/processor/test_blocking.py +++ b/tests/units/reflex_base/event/processor/test_blocking.py @@ -138,15 +138,19 @@ async def test_blocking_just_over_threshold_is_detected( assert "state.blocking" in caplog_blocking.text -async def test_blocking_outside_event_task_is_not_reported( +async def test_blocking_outside_event_task_is_reported_with_task_name( caplog_blocking: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch ): - """Only report tasks that process an event.""" + """Report blocking in a task that processes no event by the task name.""" probe_queued = _track_queued_probe(asyncio.get_running_loop(), monkeypatch) await _run_monitored(_blocking(probe_queued, 0.12), "other_task") - assert "the event loop" not in caplog_blocking.text + assert "Task 'other_task' is blocking the event loop" in caplog_blocking.text + assert "Task 'other_task' blocked the event loop for 0.1" in caplog_blocking.text + assert "time.sleep(duration) # noqa: ASYNC251" in caplog_blocking.text + assert "Event handler" not in caplog_blocking.text + assert "computed vars" not in caplog_blocking.text async def test_awaiting_event_is_not_reported( diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index 85045933672..b009404fe8d 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -9,6 +9,7 @@ import threading import time from collections.abc import AsyncIterator, Mapping +from pathlib import Path from typing import Any from unittest.mock import Mock @@ -696,10 +697,10 @@ async def test_processor_reports_innermost_app_frame( async with mock_event_processor as ep: await ep.enqueue(token, Event.from_event_type(blocking_stdlib_event())[0]) - assert ( - "is blocking the event loop at " - "tests/units/reflex_base/event/processor/test_event_processor.py:" - ) in caplog.text + path = Path( + "tests", "units", "reflex_base", "event", "processor", "test_event_processor.py" + ) + assert f"is blocking the event loop at {path}:" in caplog.text assert "in _blocking_stdlib_handler (threshold" in caplog.text