diff --git a/news/6744.performance.md b/news/6744.performance.md new file mode 100644 index 00000000000..857ba366dd8 --- /dev/null +++ b/news/6744.performance.md @@ -0,0 +1 @@ +Reduce per-event websocket overhead: client headers are decoded once per connection instead of per event, route resolution is cached, and chained-event routing no longer runs while holding the state lock. diff --git a/packages/reflex-base/news/6744.performance.md b/packages/reflex-base/news/6744.performance.md new file mode 100644 index 00000000000..cedfa59b1b9 --- /dev/null +++ b/packages/reflex-base/news/6744.performance.md @@ -0,0 +1 @@ +Event handlers' final chained-event routing now runs after the state lock is released, shortening lock hold time per event; deltas and streamed (yielded) updates still emit immediately. diff --git a/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py b/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py index f361b2f3969..8e7f0954f21 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py @@ -6,7 +6,7 @@ import functools import inspect import warnings -from collections.abc import Mapping, Sequence +from collections.abc import Awaitable, Callable, Mapping, Sequence from enum import Enum from importlib.util import find_spec from typing import TYPE_CHECKING, Any @@ -190,6 +190,7 @@ async def chain_updates( events: EventSpec | list[EventSpec] | None, handler_name: str, root_state: BaseState | None = None, + deferred: list[Callable[[], Awaitable[None]]] | None = None, ) -> None: """Chain yielded events and emit a delta to the frontend. @@ -200,6 +201,12 @@ async def chain_updates( events: The events to queue with the update. handler_name: The name of the handler that yielded the events, used for error messages. root_state: The root state of the app, no delta emitted if omitted. + deferred: If provided, the routing of the chained events is appended + here instead of being awaited, so the caller can run it after + releasing the state lock. The delta is always emitted inline: its + values reference live state containers (``get_value`` unwraps + proxies without copying), so it must be serialized before another + lock holder can mutate them. """ from reflex.event import Event @@ -215,10 +222,14 @@ async def chain_updates( root_state._clean() # Convert valid EventHandler and EventSpec into Event - if fixed_events := Event.from_event_type( + fixed_events = Event.from_event_type( _check_valid_yield(events, handler_name=handler_name), - ): - await _route_events(ctx, fixed_events) + ) + if fixed_events: + if deferred is not None: + deferred.append(functools.partial(_route_events, ctx, fixed_events)) + else: + await _route_events(ctx, fixed_events) async def process_event( @@ -226,6 +237,7 @@ async def process_event( payload: dict, state: BaseState | StateProxy, root_state: BaseState, + deferred: list[Callable[[], Awaitable[None]]] | None = None, ): """Process event. @@ -234,6 +246,10 @@ async def process_event( payload: The event payload. state: State to process the handler. root_state: The root state of the app, used for emitting deltas. + deferred: If provided, routing of the handler's final chained events + is appended here instead of being awaited, so the caller can run + it after releasing the state lock. Deltas are always emitted + inline. Raises: ValueError: If a string value is received for an int or float type and cannot be converted. @@ -263,7 +279,9 @@ async def process_event( if inspect.isasyncgen(events): async for event in events: await chain_updates(event, root_state=root_state, handler_name=handler_name) - await chain_updates(None, root_state=root_state, handler_name=handler_name) + await chain_updates( + None, root_state=root_state, handler_name=handler_name, deferred=deferred + ) # Handle regular generators. elif inspect.isgenerator(events): @@ -277,13 +295,20 @@ async def process_event( # in the loop, we must catch StopIteration to access it if si.value is not None: await chain_updates( - si.value, root_state=root_state, handler_name=handler_name + si.value, + root_state=root_state, + handler_name=handler_name, + deferred=deferred, ) - await chain_updates(None, root_state=root_state, handler_name=handler_name) + await chain_updates( + None, root_state=root_state, handler_name=handler_name, deferred=deferred + ) # Handle regular event chains. else: - await chain_updates(events, root_state=root_state, handler_name=handler_name) + await chain_updates( + events, root_state=root_state, handler_name=handler_name, deferred=deferred + ) class BaseStateEventProcessor(EventProcessor): @@ -336,6 +361,8 @@ async def _execute_event( ctx = entry.ctx event = entry.event router_data = event.router_data or {} + deferred_emits: list[Callable[[], Awaitable[None]]] | None = None + background_states: tuple[BaseState, BaseState] | None = None # Get the state for the session exclusively. async with ctx.state_manager.modify_state_with_links( BaseStateToken( @@ -377,22 +404,34 @@ async def _execute_event( if needs_to_rehydrate: await self._rehydrate(root_state) - # Process non-background events while holding the lock. + # Process non-background events while holding the lock. The + # routing of any final chained events is collected and run after + # the lock is released, shortening the hold time; deltas are + # emitted inline since their values reference live state data. if not registered_handler.handler.is_background: + deferred_emits = [] await process_event( handler=registered_handler.handler, payload=event.payload, state=substate, root_state=root_state, + deferred=deferred_emits, ) - return - # Otherwise drop the state lock and start processing the background task with a proxy state. - await process_event( - handler=registered_handler.handler, - state=StateProxy(substate), - payload=event.payload, - root_state=root_state, - ) + else: + background_states = (substate, root_state) + if deferred_emits is not None: + for emit in deferred_emits: + await emit() + elif background_states is not None: + # The state lock was dropped above; process the background task + # with a proxy state. + substate, root_state = background_states + await process_event( + handler=registered_handler.handler, + state=StateProxy(substate), + payload=event.payload, + root_state=root_state, + ) async def _handle_backend_exception( self, ex: Exception, ev_ctx: EventContext | None = None diff --git a/reflex/app.py b/reflex/app.py index 9364c547436..74faf7f269b 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -1618,6 +1618,9 @@ def __init__(self, namespace: str, app: App): # Use TokenManager for distributed duplicate tab prevention self._token_manager = TokenManager.create() + # Decoded headers and client IP per sid; fixed for a connection's lifetime. + self._sid_client_info: dict[str, tuple[dict[str, str], str]] = {} + @property def token_to_sid(self) -> Mapping[str, str]: """Get token to SID mapping for backward compatibility. @@ -1672,6 +1675,7 @@ def on_disconnect(self, sid: str) -> asyncio.Task | None: Returns: An asyncio Task for cleaning up the token, or None. """ + self._sid_client_info.pop(sid, None) # Get token before cleaning up disconnect_token = self.sid_to_token.get(sid) if disconnect_token: @@ -1769,29 +1773,30 @@ async def on_event(self, sid: str, data: Any): msg = "Socket.IO environ is not initialized." raise RuntimeError(msg) - # Get the client headers. - headers = { - k.decode("utf-8"): v.decode("utf-8") - for (k, v) in environ["asgi.scope"]["headers"] - } - - # Get the client IP - try: - client_ip = environ["asgi.scope"]["client"][0] - headers["asgi-scope-client"] = client_ip - except (KeyError, IndexError): - client_ip = environ.get("REMOTE_ADDR", "0.0.0.0") - - # Unroll reverse proxy forwarded headers. - client_ip = ( - headers - .get( - "x-forwarded-for", - client_ip, + # Get the client headers and IP, decoded once per connection since + # they are fixed for the lifetime of a sid. + client_info = self._sid_client_info.get(sid) + if client_info is None: + cached_headers = { + k.decode("utf-8"): v.decode("utf-8") + for (k, v) in environ["asgi.scope"]["headers"] + } + try: + client_ip = environ["asgi.scope"]["client"][0] + cached_headers["asgi-scope-client"] = client_ip + except (KeyError, IndexError): + client_ip = environ.get("REMOTE_ADDR", "0.0.0.0") + # Unroll reverse proxy forwarded headers. + client_ip = ( + cached_headers + .get("x-forwarded-for", client_ip) + .partition(",")[0] + .strip() ) - .partition(",")[0] - .strip() - ) + client_info = self._sid_client_info[sid] = (cached_headers, client_ip) + # Copy so downstream router_data mutations cannot leak into the cache. + headers = dict(client_info[0]) + client_ip = client_info[1] router_data = event.router_data router_data.update({ constants.RouteVar.QUERY: format.format_query_params(event.router_data), diff --git a/reflex/route.py b/reflex/route.py index 4b450fb6e25..b9de8cb18f9 100644 --- a/reflex/route.py +++ b/reflex/route.py @@ -2,6 +2,7 @@ from __future__ import annotations +import functools import re from collections.abc import Callable @@ -203,6 +204,10 @@ def get_router(routes: list[str]) -> Callable[[str], str | None]: for keyworded_route, original_route in sorted_routes_by_specificity ] + # Cached per router: the route set is fixed for the router's lifetime and + # clients mostly re-send the same path, so the linear regex scan runs once + # per distinct path instead of on every event. + @functools.lru_cache(maxsize=1024) def get_route(path: str) -> str | None: """Get the first matching route for a given path.