diff --git a/tests/benchmarks/README.md b/tests/benchmarks/README.md new file mode 100644 index 00000000000..2ec666ec02b --- /dev/null +++ b/tests/benchmarks/README.md @@ -0,0 +1,16 @@ +# Event processing benchmarks + +```sh +uv run pytest tests/benchmarks/test_event_processing.py --codspeed +``` + +Omit `--codspeed` for local wall-clock measurements. + +- `counter`: three increments 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; each batch returns to the same filter and +sort direction. 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..96639a70b79 100644 --- a/tests/benchmarks/test_event_processing.py +++ b/tests/benchmarks/test_event_processing.py @@ -1,9 +1,4 @@ -"""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 traceback @@ -17,31 +12,52 @@ 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 +from .fixtures import BenchmarkState, TableState -@pytest_asyncio.fixture -async def event_processing_harness(): - """Set up the full event processing pipeline for benchmarking. +@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. - Creates a ``BaseStateEventProcessor`` wired to a real - ``StateManagerMemory`` with mock emit callbacks. Events are - enqueued directly and deltas are collected via the emit callback. + Args: + request: Selects the counter or table workload. 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]]]] = [] + table = request.param == "table" + handler = ( + TableState.event_handlers["set_status"] + if table + else BenchmarkState.event_handlers["increment"] + ) + payloads = ( + [{"status": status} for status in ("open", "", "paid") * 2] + if table + else [{}] * 3 + ) + events = [ + Event( + name=format_event_handler(handler), + router_data={"query": {}, "path": "/"}, + payload=payload, + ) + for payload in payloads + ] + 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 + if table: + json_dumps(StateUpdate(delta=delta), separators=(",", ":")) async def emit_event_impl(token: str, *events: Event) -> None: pass @@ -54,69 +70,45 @@ 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 - - yield run_events - - await state_manager.close() + assert emitted == len(events) + try: + if table: + await run_events() + yield run_events + finally: + await state_manager.close() -def test_process_event( - event_processing_harness, - benchmark: BenchmarkFixture, -): - """Benchmark processing 3 increment events through the full pipeline. - The first event creates fresh state (cold path), the next two reuse - the existing state (warm path). Only event processing is timed. +def test_process_event(event_processing_harness, benchmark: BenchmarkFixture): + """Benchmark a batch of three 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())