diff --git a/tests/benchmarks/README.md b/tests/benchmarks/README.md new file mode 100644 index 00000000000..379ca638fd5 --- /dev/null +++ b/tests/benchmarks/README.md @@ -0,0 +1,20 @@ +# Event processing benchmarks + +```sh +uv run pytest tests/benchmarks/test_event_processing.py --codspeed +``` + +Omit `--codspeed` for local wall-clock measurements. + +- `counter`: two increments and two decrements through the in-memory event + processor. +- `table`: six filter/sort events over 1,000 dataclass orders, including computed + rows and totals and JSON encoding of each `StateUpdate`. The batch cycles + through open/all/paid orders twice, reversing sort direction on every event. + +The table is warmed before timing, and both batches return state to its +starting point so every sample measures identical work. +`test_table_event_deltas` separately verifies the serialized rows, sort +direction, and totals that the table benchmark encodes. +Timing includes processor startup/shutdown, but excludes initial +hydration, Socket.IO packet framing, network transport, databases, and rendering. diff --git a/tests/benchmarks/fixtures.py b/tests/benchmarks/fixtures.py index 63469330109..f5afda90746 100644 --- a/tests/benchmarks/fixtures.py +++ b/tests/benchmarks/fixtures.py @@ -272,6 +272,69 @@ def enter_component( return enter_component +@dataclass +class Order: + """An order in the table event benchmark.""" + + name: str + customer: str + amount: float + status: str + + +class TableState(rx.State): + """A 1000-row table with filtering, sorting, and a computed total.""" + + orders: rx.Field[list[Order]] = rx.field( + default_factory=lambda: [ + Order( + name=f"order {i}", + customer=f"customer {i % 50}", + amount=i * 1.5, + status=("open", "paid", "shipped")[i % 3], + ) + for i in range(1000) + ] + ) + status: rx.Field[str] = rx.field("") + sort_reverse: rx.Field[bool] = rx.field(False) + + @rx.event + def set_status(self, status: str): + """Filter the table by status, flipping the sort direction. + + Args: + status: The status to keep, or an empty string for all rows. + """ + self.status = status + self.sort_reverse = not self.sort_reverse + + @rx.var + def filtered_orders(self) -> list[Order]: + """The rows matching the filter, sorted. + + Returns: + The filtered, sorted rows. + """ + orders = self.orders + if self.status: + orders = [order for order in orders if order.status == self.status] + return sorted( + orders, + key=lambda order: order.amount, + reverse=self.sort_reverse, + ) + + @rx.var + def total_amount(self) -> float: + """The amount summed over the filtered rows. + + Returns: + The total amount. + """ + return sum(order.amount for order in self.filtered_orders) + + class BenchmarkState(rx.State): """State for the benchmark.""" diff --git a/tests/benchmarks/test_event_processing.py b/tests/benchmarks/test_event_processing.py index 15acf8094d4..c4fb3e13d95 100644 --- a/tests/benchmarks/test_event_processing.py +++ b/tests/benchmarks/test_event_processing.py @@ -1,47 +1,105 @@ -"""Benchmark for the event processing pipeline. - -Measures the time from enqueuing events via ``BaseStateEventProcessor`` -to collecting all emitted ``StateUpdate`` deltas, with mock emit -callbacks that record the deltas. -""" +"""Benchmark counter and table events through the in-memory event pipeline.""" import asyncio +import json import traceback -from collections.abc import Mapping +from collections.abc import AsyncIterator, Awaitable, Callable, Mapping +from contextlib import asynccontextmanager from typing import Any from unittest import mock import pytest import pytest_asyncio from pytest_codspeed import BenchmarkFixture +from reflex_base.constants.state import FIELD_MARKER from reflex_base.event import Event from reflex_base.event.context import EventContext from reflex_base.event.processor import BaseStateEventProcessor -from reflex_base.utils.format import format_event_handler +from reflex_base.utils.format import format_event_handler, json_dumps from reflex.istate.manager.memory import StateManagerMemory +from reflex.state import StateUpdate + +from .fixtures import BenchmarkState, TableState + +TABLE_STATUSES = ("open", "", "paid") + + +def _encode_delta(delta: Mapping[str, Mapping[str, Any]]) -> str: + """Serialize a delta the way the socket path does. + + Args: + delta: The state changes emitted by the processor. + + Returns: + The delta encoded as a StateUpdate envelope. + """ + return json_dumps(StateUpdate(delta=delta), separators=(",", ":")) + -from .fixtures import BenchmarkState +def _events(handler_name: str, payloads: list[dict[str, Any]]) -> list[Event]: + """Build one event per payload for the given handler. + Args: + handler_name: The formatted event handler name. + payloads: One payload per event. + + Returns: + The events to enqueue. + """ + return [ + Event( + name=handler_name, + router_data={"query": {}, "path": "/"}, + payload=payload, + ) + for payload in payloads + ] -@pytest_asyncio.fixture -async def event_processing_harness(): - """Set up the full event processing pipeline for benchmarking. - Creates a ``BaseStateEventProcessor`` wired to a real - ``StateManagerMemory`` with mock emit callbacks. Events are - enqueued directly and deltas are collected via the emit callback. +def _counter_events() -> list[Event]: + """Two increments followed by two decrements, returning to the start. + + Returns: + The counter event batch. + """ + increment = format_event_handler(BenchmarkState.event_handlers["increment"]) + decrement = format_event_handler(BenchmarkState.event_handlers["decrement"]) + return _events(increment, [{}] * 2) + _events(decrement, [{}] * 2) + + +def _table_events() -> list[Event]: + """Two filter cycles, ending on the starting sort direction. + + Returns: + The table event batch. + """ + set_status = format_event_handler(TableState.event_handlers["set_status"]) + return _events(set_status, [{"status": status} for status in TABLE_STATUSES * 2]) + + +@asynccontextmanager +async def _event_pipeline( + events: list[Event], + on_delta: Callable[[Mapping[str, Mapping[str, Any]]], Any], +) -> AsyncIterator[Callable[[], Awaitable[None]]]: + """Wire a ``BaseStateEventProcessor`` to an in-memory state manager. + + Args: + events: The batch to enqueue on each run. + on_delta: Called with each emitted delta. Yields: - An async callable that enqueues the given number of events - and waits for all expected deltas. + An async callable that processes one batch and checks its delta count. """ - emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]] = [] + emitted = 0 async def emit_delta_impl( # noqa: RUF029 token: str, delta: Mapping[str, Mapping[str, Any]] ) -> None: - emitted_deltas.append((token, delta)) + nonlocal emitted + emitted += 1 + on_delta(delta) async def emit_event_impl(token: str, *events: Event) -> None: pass @@ -54,69 +112,98 @@ def handle_backend_exception(ex: Exception) -> None: backend_exception_handler=handle_backend_exception, graceful_shutdown_timeout=5, ) - # Mock _rehydrate so the processor doesn't try to push full state - # to a non-existent frontend on the first event. + # Skip initial hydration because there is no frontend. with mock.patch.object(processor, "_rehydrate", new=mock.AsyncMock()): state_manager = StateManagerMemory() - root_context = EventContext( + processor._root_context = EventContext( token="", state_manager=state_manager, enqueue_impl=processor.enqueue_many, emit_delta_impl=emit_delta_impl, emit_event_impl=emit_event_impl, ) - processor._root_context = root_context - - token = "benchmark-token" - handler_name = format_event_handler(BenchmarkState.event_handlers["increment"]) - event = Event( - name=handler_name, - router_data={ - "query": {}, - "path": "/", - }, - payload={}, - ) - - async def run_events(num_events: int, num_expected_deltas: int) -> None: - """Enqueue events and wait for all deltas to be emitted. - - Args: - num_events: Number of increment events to enqueue. - num_expected_deltas: How many deltas to wait for. - """ - emitted_deltas.clear() + async def run_events() -> None: + """Process the batch and verify that each event emitted a delta.""" + nonlocal emitted + emitted = 0 async with processor as p: async for _ in asyncio.as_completed([ - await p.enqueue(token, event) for _ in range(num_events) + await p.enqueue("benchmark-token", event) for event in events ]): pass - assert len(emitted_deltas) == num_expected_deltas + assert emitted == len(events) - yield run_events + try: + yield run_events + finally: + await state_manager.close() - await state_manager.close() +@pytest_asyncio.fixture(params=["counter", "table"]) +async def event_processing_harness(request: pytest.FixtureRequest): + """Set up a fixed event batch, warming the table before timing. -def test_process_event( - event_processing_harness, - benchmark: BenchmarkFixture, -): - """Benchmark processing 3 increment events through the full pipeline. + Both batches return state to its starting point, so every benchmark + sample measures identical work. - The first event creates fresh state (cold path), the next two reuse - the existing state (warm path). Only event processing is timed. + Args: + request: Selects the counter or table workload. + + Yields: + An async callable that processes one batch and checks its delta count. + """ + table = request.param == "table" + events = _table_events() if table else _counter_events() + on_delta = _encode_delta if table else (lambda delta: None) + async with _event_pipeline(events, on_delta) as run_events: + if table: + await run_events() + yield run_events + + +def test_process_event(event_processing_harness, benchmark: BenchmarkFixture): + """Benchmark a batch of four counter events or six table events. Args: - event_processing_harness: The run_events async callable. + event_processing_harness: The async batch runner. benchmark: The codspeed benchmark fixture. """ - run_events = event_processing_harness loop = asyncio.get_event_loop() - # Each event handler (increment) does a single state mutation with - # no yields, so we expect 1 delta per event = 3 total. @benchmark def _(): - loop.run_until_complete(run_events(num_events=3, num_expected_deltas=3)) + loop.run_until_complete(event_processing_harness()) + + +@pytest.mark.asyncio +async def test_table_event_deltas(): + """Verify filtered rows, sort direction, and totals across repeated batches.""" + updates: list[str] = [] + + async with _event_pipeline( + _table_events(), lambda delta: updates.append(_encode_delta(delta)) + ) as run_events: + for _ in range(2): + await run_events() + + for index, update in enumerate(updates): + status = TABLE_STATUSES[index % len(TABLE_STATUSES)] + reverse = index % 2 == 0 + expected = [ + { + "name": f"order {i}", + "customer": f"customer {i % 50}", + "amount": i * 1.5, + "status": ("open", "paid", "shipped")[i % 3], + } + for i in sorted(range(1000), reverse=reverse) + if not status or ("open", "paid", "shipped")[i % 3] == status + ] + delta = json.loads(update)["delta"][TableState.get_full_name()] + assert delta["status" + FIELD_MARKER] == status + assert delta["sort_reverse" + FIELD_MARKER] == reverse + assert delta["filtered_orders" + FIELD_MARKER] == expected + assert delta["total_amount" + FIELD_MARKER] == sum( + row["amount"] for row in expected + )