Skip to content

Commit d82c422

Browse files
perf: cut per-event CPU on the runtime hot paths (#7370)
* perf: cut per-event CPU on the runtime hot paths Every backend event pays a fixed cost in the event processor and a per-value cost in the state machinery: reading fields and computed vars, wrapping values in mutable proxies, building the delta, encoding it to JSON and, with a redis or disk manager, pickling the touched states. Each of those paths did work per call that only has to happen once per class, per handler or per type, and a few did work that was never needed at all. Computed vars: a cache hit re-validated the whole cached value against the return type on every read (a 1000-element cached list cost ~1 ms per read; now ~0.2 us), the three cache attribute names were formatted per access, the last-updated timestamp was stored for every var although only interval vars read it, and marking a var dirty walked dependency dicts for names that have no dependents. The type mismatch is now logged once per computed value. Fields: scalar values skip the mutability check, and the dirty marking does not re-assign an already set touched flag through the dev-mode setattr hook. Mutable proxies: a proxy built for a list element no longer goes through the class lookup and the type-call dispatch, immutable values return before any tracking check in __getattr__, __getitem__ and __iter__, the dataclass frame check only runs for proxied dataclasses (via sys._getframe instead of a five-frame walk), and a method that is rebound to the proxy is no longer wrapped in a FunctionWrapper first. Event processor: the event context is built once per enqueue (fork takes the router_data), transaction ids come straight from os.urandom, the handler's coroutine-ness, background and supersedes flags and type hints are resolved once per handler, the per-event imports are hoisted, a handler that yields nothing skips the yield validation, the memory manager's modify_state and modify_state_with_links are plain async context managers instead of async generators, substate lookup resolves the class path once, and build_delta reads field values without constructing a proxy to unwrap. Serialization: json_dumps shares one encoder per set of separators instead of building a JSONEncoder per call, dataclasses serialize through a per-type serializer instead of dataclasses.fields() per instance, proxied values resolve their serializer once instead of dispatching twice, and StateUpdate serializes from a precomputed field tuple. State managers: redis persists every touched state of a tree in one pipeline instead of one write task (and one lock check) per state, deserializing a state updates the instance dict and drops the stale keys instead of filtering per key, the state-size check only runs on states over the limit, and the disk manager only derives a token for the states it writes. Measured on an idle 4-core VM (min of interleaved rounds): - in-process event pipeline, counter handler with 4 computed vars: 266 -> 205 us/event; 1000-row filtered/sorted table: 8.5 -> 4.4 ms/event - proxy iteration of 1000 dataclass rows: -55%; sorted(): -62%; attribute chains: -25%; write paths unchanged - cached computed var hit: 1.32 -> 0.19 us (scalar), 992 -> 0.19 us (list of 1000); scalar field read -35%, write -36% - JSON: 4-var delta 4.5 -> 2.7 us; 1000 proxied dataclass rows 2.5 -> 1.6 ms - redis manager, 35-state tree: 8.1 -> 2.1 ms/event; pickle 5.5 -> 3.3 us Claude-Session: https://claude.ai/code/session_01DKCRZTdcqM4Smu9MKrjfhx * perf: cut the per-operation cost of building Vars Every derived Var a page builds (arithmetic, comparisons, string and array methods, rx.cond, rx.foreach) goes through LiteralVar.create for its operands, the var_operation wrapper, and Var.guess_type on the result. Each of those did work per call that only depends on the type involved, or imported a module on every call: - LiteralVar.create ran two relative imports before looking at the value; the exact-type lookup now comes first and the imports (which register the literal classes) only run on a miss. - Var.guess_type resolved the type alias, unwrapped Optional, tested the origin and union, ran two registry caches and the whole Var.to dispatch on every call; a plain cast now comes from a per-type cache that the registry resets when a Var subclass is added. - Var.to imported ObjectVar and ran the serializer probe before consulting the registry for a Var-class target. - ToOperation.__getattr__ imported ObjectVar to answer _js_expr, which every operand interpolation reads. - The var_operation wrapper built two dicts, two item lists and a concatenation per call to pass a tuple of pairs along. - Each cached Var property added one __del__ link that raised and caught AttributeError for every property never read; the owner now has one __del__ that probes the instance dict for all of its cached fields. - A color name made of plain literals built six Vars and a concat operation for a string that is known at compile time. The generated code is unchanged (124 snapshots of a Var corpus and the compiled benchmark pages are identical). Measured, min of interleaved rounds on an idle core: arithmetic/comparison/string/array operations -47 to -55%, match -35%, iteration -28%; a var-heavy page evaluates -29%, the stateful page -17%, a full page compile -7 to -11%. Claude-Session: https://claude.ai/code/session_01DKCRZTdcqM4Smu9MKrjfhx * Assert which states the batched redis write stores The batched set_state tests only counted pipeline constructions, so a batch that also wrote an untouched state, or a direct write for an untouched tree, still passed. Check the stored keys of the tree: only the touched states are written, and an untouched tree writes nothing at all. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Check the redis lock atomically with the batched state write The batched set_state checked the lock once, before serializing the whole tree, where the per-state writes it replaced each re-checked it before their own write. A tree whose serialization outlasted the lock could then overwrite the writes of the next lock holder. A held lock now writes the touched states through a script that compares the lock id and sets the keys atomically, still in one round trip (and cheaper than the MULTI/EXEC pipeline), raising LockExpiredError when the lock changed hands. The mock redis runs the script for the unit tests. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Cache a computed var whose value is dataclasses.MISSING The cached computed var read used dataclasses.MISSING to mean "no cached value", so a getter returning MISSING ran again, and marked the state touched again, on every read. Use a private sentinel in the sync and async paths. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Release the state lock when linking the state fails modify_state_with_links released the lock only for errors raised while entering the linked states: an error from the link lookup or the shared-state imports, after the lock was taken, left the token locked. Guard the whole link setup. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Cache proxy classes for dataclass types only _proxy_class cached its answer for every mutable type it saw, so it kept each one alive for the life of the process, including model classes an app generates at runtime, although any non-dataclass type is proxied through the base class. Check for a dataclass first and cache only the classes generated for dataclass types, as before the cache was introduced. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Read get_value through a StateProxy on the state it wraps get_value took its field fast path for a StateProxy too, since the proxy reports the wrapped state's class, and Field._get_raw then looked up the proxy's parent_state, which a StateProxy refuses outside `async with self`. Reading a value through a background task's proxy raised ImmutableStateError. Take the fast path for a state itself only; a proxy reads through getattr, as before. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Reuse the empty type hints of an unannotated event handler process_event fell back to resolving a handler's type hints whenever its cached hints were falsy, which includes the empty mapping cached for a handler without annotations, so those handlers resolved their hints again on every event. Fall back only when the hints were not resolvable at registration (None). Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Shorten the runtime performance news fragments Say what got faster for users in a sentence or two, keeping the two behavior notes (the computed var type log and the once-per-handler is_background/supersedes reads), and leave the implementation narrative to the PR. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Fix type errors in the lock-checked write and its tests Cast the script's untyped reply to the lock id it carries, as _lock does for GETDEL, and patch reflex.utils.types directly in the type hints test instead of reaching it through the processor module, which does not export it. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE * Type the dataclass proxy-class cache for the minimum dependency versions At the declared wrapt floor (1.17.0, which ships no type information) pyright resolves wrapt.ObjectProxy to Unknown, so it cannot prove type[MutableProxy] hashable, and functools.cache's Hashable parameters rejected the base proxy class: check-min-deps failed for reflex. Keep the classes generated for dataclass types in a dict per base proxy class, keyed by the dataclass type, instead. Other types are still never cached, and a cached lookup no longer builds and hashes a key tuple. Claude-Session: https://claude.ai/code/session_01PXcFKAgnzhgdnSvhuCmxWE --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 8ee5d41 commit d82c422

26 files changed

Lines changed: 1665 additions & 310 deletions
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Process events with less CPU on the backend: iterating, sorting and reading list and dataclass state values costs 25–75% less, and the redis state manager writes the changed states of a session in one round trip.
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Lower the CPU cost of every backend event: cached computed var reads, field reads and writes, event dispatch and the JSON encoding of state deltas are faster. A computed var returning a value of the wrong type is now logged once per computed value instead of on every read, and `EventHandler.is_background` and `EventHandler.supersedes` are read once per handler, so mark the function before the handler is first used.
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Building Vars (operations, comparisons, `rx.cond`, `rx.foreach`) is up to twice as fast, so pages that derive many Vars compile faster; the generated code is unchanged.

‎packages/reflex-base/src/reflex_base/event/__init__.py‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
import warnings
1010
from base64 import b64encode
1111
from collections.abc import Callable, Mapping, Sequence
12-
from functools import lru_cache, partial
12+
from functools import cached_property, lru_cache, partial
1313
from typing import (
1414
TYPE_CHECKING,
1515
Annotated,
@@ -611,16 +611,27 @@ def __class_getitem__(cls, args_spec: str) -> Annotated:
611611
"""
612612
return Annotated[cls, args_spec]
613613

614-
@property
614+
@cached_property
615615
def is_background(self) -> bool:
616616
"""Whether the event handler is a background task.
617617
618+
Read on the first access, so the function has to be marked before then.
619+
618620
Returns:
619621
True if the event handler is marked as a background task.
620622
"""
621623
return getattr(self.fn, BACKGROUND_TASK_MARKER, False)
622624

623-
@property
625+
@cached_property
626+
def _is_coroutine_function(self) -> bool:
627+
"""Whether the handler function is a coroutine function.
628+
629+
Returns:
630+
True if calling the handler function returns a coroutine.
631+
"""
632+
return inspect.iscoroutinefunction(self.fn)
633+
634+
@cached_property
624635
def supersedes(self) -> bool:
625636
"""Whether a newer invocation supersedes an older one.
626637
@@ -636,6 +647,7 @@ def supersedes(self) -> bool:
636647
Cancellation is cooperative: a handler that never yields to the event
637648
loop runs to completion, and only its not-yet-started chained events
638649
are skipped.
650+
Read on the first access, so the function has to be marked before then.
639651
640652
Returns:
641653
True if the event handler is marked as superseding.

‎packages/reflex-base/src/reflex_base/event/context.py‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
import dataclasses
66
import functools
7-
import uuid
7+
import os
88
from collections.abc import Callable, Mapping
99
from typing import TYPE_CHECKING, Any, Protocol
1010

@@ -31,6 +31,15 @@ def get_name(cls: type | Callable) -> str:
3131
return to_snake_case(f"{module}___{qualname}")
3232

3333

34+
def _new_txid() -> str:
35+
"""Generate a transaction id.
36+
37+
Returns:
38+
12 random hex digits, the same 48 random bits as ``uuid4().hex[:12]``.
39+
"""
40+
return os.urandom(6).hex()
41+
42+
3443
class EnqueueProtocol(Protocol):
3544
"""Protocol for the enqueue function in the event context."""
3645

@@ -89,7 +98,7 @@ class EventContext(BaseContext):
8998

9099
# Each event is associated with a top-level transaction id. Empty for a
91100
# context that belongs to no event, like an event processor's root context.
92-
txid: str = dataclasses.field(default_factory=lambda: uuid.uuid4().hex[:12])
101+
txid: str = dataclasses.field(default_factory=_new_txid)
93102
# The txid of another EventContext that enqueued this context's event, or
94103
# None for a top-level event.
95104
parent_txid: str | None = None
@@ -110,11 +119,14 @@ class EventContext(BaseContext):
110119
# event a handler yields resolves against the view that produced it.
111120
router_data: dict[str, Any] = dataclasses.field(default_factory=dict, repr=False)
112121

113-
def fork(self, token: str | None = None) -> EventContext:
122+
def fork(
123+
self, token: str | None = None, router_data: dict[str, Any] | None = None
124+
) -> EventContext:
114125
"""Return a new EventContext with the specified fields replaced.
115126
116127
Args:
117128
token: The client token for the new context.
129+
router_data: The routing data for the new context, inherited from this context if empty.
118130
119131
Returns:
120132
A new EventContext with the specified fields replaced.
@@ -127,7 +139,7 @@ def fork(self, token: str | None = None) -> EventContext:
127139
enqueue_impl=self.enqueue_impl,
128140
emit_delta_impl=self.emit_delta_impl,
129141
emit_event_impl=self.emit_event_impl,
130-
router_data=self.router_data,
142+
router_data=router_data or self.router_data,
131143
otel_context=otel.capture_context(),
132144
)
133145

‎packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py‎

Lines changed: 24 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -18,16 +18,17 @@
1818
from reflex.utils import types
1919
from reflex_base import otel
2020
from reflex_base.constants import CompileVars
21+
from reflex_base.event import Event, EventHandler, EventSpec
2122
from reflex_base.event.context import EventContext
2223
from reflex_base.event.processor.event_processor import EventProcessor, EventQueueEntry
2324
from reflex_base.event.processor.future import EventFuture
2425
from reflex_base.registry import RegisteredEventHandler
2526
from reflex_base.utils.format import format_event_handler
27+
from reflex_base.utils.serializers import deserializers
2628

2729
logger = logging.getLogger(__name__)
2830

2931
if TYPE_CHECKING:
30-
from reflex.event import Event, EventHandler
3132
from reflex.state import BaseState
3233

3334
# Resolved once at import: find_spec on a missing package scans sys.path (~90us),
@@ -76,7 +77,6 @@ def _check_valid_yield(events: Any, handler_name: str = "unknown") -> Any:
7677
Raises:
7778
TypeError: If any of the events are not valid.
7879
"""
79-
from reflex.event import Event, EventHandler, EventSpec
8080

8181
def _is_valid_type(events: Any) -> bool:
8282
return isinstance(events, (Event, EventHandler, EventSpec))
@@ -121,9 +121,6 @@ def _transform_event_arg(value: Any, hinted_args: Any) -> Any:
121121
Raises:
122122
ValueError: If a string value is received for an int or float type and cannot be converted.
123123
"""
124-
from reflex.model import Model
125-
from reflex.utils.serializers import deserializers
126-
127124
if hinted_args is Any:
128125
return value
129126
if types.is_union(hinted_args):
@@ -135,6 +132,8 @@ def _transform_event_arg(value: Any, hinted_args: Any) -> Any:
135132
and isinstance(hinted_args, type)
136133
and not types.is_generic_alias(hinted_args) # py3.10
137134
):
135+
from reflex.model import Model
136+
138137
if issubclass(hinted_args, Model):
139138
# Remove non-fields from the payload
140139
return hinted_args(**{
@@ -181,7 +180,7 @@ def _transform_event_payload(
181180
The transformed event payload.
182181
"""
183182
transformed = {}
184-
for arg, value in list(payload.items()):
183+
for arg, value in payload.items():
185184
hinted_args = type_hints.get(arg, Any)
186185
try:
187186
transformed[arg] = _transform_event_arg(value, hinted_args)
@@ -231,8 +230,6 @@ async def chain_updates(
231230
handler_name: The name of the handler that yielded the events, used for error messages.
232231
root_state: The root state of the app, no delta emitted if omitted.
233232
"""
234-
from reflex.event import Event
235-
236233
ctx = EventContext.get()
237234

238235
if root_state is not None:
@@ -250,8 +247,10 @@ async def chain_updates(
250247
root_state._clean()
251248

252249
# Convert valid EventHandler and EventSpec into Event
253-
if fixed_events := Event.from_event_type(
254-
_check_valid_yield(events, handler_name=handler_name),
250+
if events is not None and (
251+
fixed_events := Event.from_event_type(
252+
_check_valid_yield(events, handler_name=handler_name),
253+
)
255254
):
256255
await _route_events(ctx, fixed_events)
257256

@@ -305,13 +304,16 @@ async def process_event(
305304
Raises:
306305
ValueError: If a string value is received for an int or float type and cannot be converted.
307306
"""
308-
handler_name = handler.fn.__qualname__
309-
310-
# Get the function to process the event.
311-
fn = functools.partial(handler.fn, state)
307+
fn = handler.fn
308+
handler_name = fn.__qualname__
312309

313310
try:
314-
type_hints = types.get_type_hints(handler.fn)
311+
# Resolved hints are cached on the handler, empty for an unannotated one;
312+
# fall back for ones that were not resolvable at registration (None),
313+
# raising again if still unresolved.
314+
type_hints = handler._type_hints
315+
if type_hints is None:
316+
type_hints = types.get_type_hints(fn)
315317
payload = _transform_event_payload(payload, type_hints)
316318
except Exception as ex:
317319
# No transformation was possible, continue with the original payload
@@ -320,12 +322,12 @@ async def process_event(
320322
)
321323

322324
# Handle async functions.
323-
if inspect.iscoroutinefunction(fn.func):
324-
events = await fn(**payload)
325+
if handler._is_coroutine_function:
326+
events = await fn(state, **payload)
325327

326328
# Handle regular functions.
327329
else:
328-
events = fn(**payload)
330+
events = fn(state, **payload)
329331
# Handle async generators.
330332
if inspect.isasyncgen(events):
331333
async for event in events:
@@ -468,16 +470,14 @@ async def _execute_event(
468470
) as state:
469471
if otel.enabled:
470472
otel.record_state_acquired(acquire_start, event)
473+
previous_router_data = state.router_data
471474
# Compatibility hack rehydrate the state before processing this event.
472475
needs_to_rehydrate = bool(
473-
not state.router_data and event.name not in _hydrate_event_names()
476+
not previous_router_data and event.name not in _hydrate_event_names()
474477
)
475478

476479
# re-assign only when the value is set and different
477-
if (
478-
router_data
479-
and (previous_router_data := state.router_data) != router_data
480-
):
480+
if router_data and previous_router_data != router_data:
481481
# only the router vars whose backing keys changed are rebuilt
482482
# and re-sent; session/headers stay put across navigations.
483483
merged_router_data = state._update_router_vars(
@@ -503,7 +503,7 @@ async def _execute_event(
503503
return
504504

505505
# Get the event's substate.
506-
substate = await state.get_state(event.state_cls)
506+
substate = await state.get_state(registered_handler.states[0])
507507
root_state = state._get_root_state()
508508

509509
if needs_to_rehydrate:

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

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -402,18 +402,22 @@ async def enqueue(
402402
already enqueued it, the returned future is already cancelled and
403403
the event is dropped.
404404
"""
405+
# The event's router_data is bound before the entry exists, so entry.ctx
406+
# is what the handler, the task metadata, and every event it yields all read.
405407
if ev_ctx is None:
406408
try:
407-
ev_ctx = EventContext.get().fork(token=token)
409+
ev_ctx = EventContext.get().fork(
410+
token=token, router_data=event.router_data
411+
)
408412
except LookupError as le:
409413
if self._root_context is not None:
410-
ev_ctx = self._root_context.fork(token=token)
414+
ev_ctx = self._root_context.fork(
415+
token=token, router_data=event.router_data
416+
)
411417
else:
412418
msg = "Event processor is not running, call .start(...) first."
413419
raise RuntimeError(msg) from le
414-
if event.router_data:
415-
# Bound before the entry exists, so entry.ctx is what the handler,
416-
# the task metadata, and every event it yields all read.
420+
elif event.router_data:
417421
ev_ctx = dataclasses.replace(ev_ctx, router_data=event.router_data)
418422
queue = self._ensure_queue_task()
419423
txid = ev_ctx.txid
@@ -579,7 +583,7 @@ def _try_clean_future(self, future: EventFuture) -> None: # type: ignore[overri
579583
# reusing the txid can chain recovery events normally.
580584
return
581585
# Not checking future.all_done() to avoid waiting for grandchildren here.
582-
if not all(c.done() for c in future.children):
586+
if (children := future.children) and not all(c.done() for c in children):
583587
return
584588
parent = future.parent
585589
self._futures.pop(future.txid, None)
@@ -761,7 +765,7 @@ def _create_event_task(
761765
coro = self._process_event_queue_entry(
762766
entry=entry, registered_handler=registered_handler
763767
)
764-
name = f"reflex_event|{entry.event.name}|{entry.ctx.token}|{time.time()}"
768+
name = f"reflex_event|{entry.event.name}|{entry.ctx.token}|{entry.ctx.txid}"
765769
loop = asyncio.get_running_loop()
766770
if (
767771
sys.version_info >= (3, 12)
@@ -934,14 +938,14 @@ def _resolve_future(
934938
True if a backend exception handler task was spawned and now owns
935939
the future's lifecycle, False otherwise.
936940
"""
937-
from reflex.utils import telemetry
938-
939941
try:
940942
result = task.result()
941943
except asyncio.CancelledError:
942944
if future is not None and not future.done():
943945
future.cancel()
944946
except Exception as ex:
947+
from reflex.utils import telemetry
948+
945949
if future is not None and not future.done():
946950
future.set_exception(ex)
947951
with contextlib.suppress(BaseException):

‎packages/reflex-base/src/reflex_base/utils/format.py‎

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -718,20 +718,43 @@ def _get_serialize() -> Callable[[Any], Any]:
718718
return _serialize
719719

720720

721-
def json_dumps(obj: Any, **kwargs) -> str:
721+
@lru_cache
722+
def _get_json_encoder(separators: tuple[str, str] | None) -> json.JSONEncoder:
723+
"""Get the JSON encoder that serializes values with the Reflex serializers.
724+
725+
``json.dumps`` builds a new encoder on every call when given any option, which
726+
dominates the cost of small payloads. Encoders keep no state between calls, so
727+
one is shared per set of separators.
728+
729+
Args:
730+
separators: The item and key separators, or None for the default ones.
731+
732+
Returns:
733+
The shared encoder.
734+
"""
735+
return json.JSONEncoder(
736+
ensure_ascii=False, default=_get_serialize(), separators=separators
737+
)
738+
739+
740+
def json_dumps(obj: Any, separators: tuple[str, str] | None = None, **kwargs) -> str:
722741
"""Takes an object and returns a jsonified string.
723742
724743
Args:
725744
obj: The object to be serialized.
745+
separators: The item and key separators, as for json.dumps.
726746
kwargs: Additional keyword arguments to pass to json.dumps.
727747
728748
Returns:
729749
A string
730750
"""
751+
if not kwargs and (separators is None or isinstance(separators, tuple)):
752+
return _get_json_encoder(separators).encode(obj)
753+
731754
kwargs.setdefault("ensure_ascii", False)
732755
kwargs.setdefault("default", _get_serialize())
733756

734-
return json.dumps(obj, **kwargs)
757+
return json.dumps(obj, separators=separators, **kwargs)
735758

736759

737760
def collect_form_dict_names(form_dict: dict[str, Any]) -> dict[str, Any]:

0 commit comments

Comments
 (0)