From d2f681a69479d5c169bdcc67dd7da17464248a50 Mon Sep 17 00:00:00 2001 From: Farhan Date: Mon, 7 Sep 2026 22:55:24 +0500 Subject: [PATCH 1/2] Add holistic table event processing benchmark --- tests/benchmarks/README.md | 26 +++ tests/benchmarks/fixtures.py | 70 ++++++++ tests/benchmarks/test_event_processing.py | 195 ++++++++++++++++++---- 3 files changed, 255 insertions(+), 36 deletions(-) create mode 100644 tests/benchmarks/README.md diff --git a/tests/benchmarks/README.md b/tests/benchmarks/README.md new file mode 100644 index 00000000000..3656815ebc1 --- /dev/null +++ b/tests/benchmarks/README.md @@ -0,0 +1,26 @@ +# Event processing benchmarks + +Run the event benchmarks with: + +```sh +uv run pytest tests/benchmarks/test_event_processing.py --codspeed +``` + +Omit `--codspeed` for local wall-clock measurements. + +`test_process_event` measures three counter increments through the event processor. +`test_process_table_event` models an order-management dashboard with 1,000 +dataclass rows. Each sample runs six events: filter open orders, clear the filter, +and filter paid orders, twice, reversing the sort direction on every event. +Each update includes sorted matching rows and a computed total. + +The table benchmark warms the state before timing and returns to the same filter +and sort direction after each sample. It covers event queueing, in-memory state +access, mutations, proxied row iteration, computed-variable invalidation and +evaluation, delta generation, and Reflex JSON serialization. Processor startup +and shutdown are included. Initial hydration, Socket.IO packet framing, network +transport, database access, and browser rendering are excluded. + +`test_table_event_deltas` checks serialized results across two batches separately +from timing, so missing rows or stale computed values cannot silently look like +performance improvements. diff --git a/tests/benchmarks/fixtures.py b/tests/benchmarks/fixtures.py index 63469330109..f417019d962 100644 --- a/tests/benchmarks/fixtures.py +++ b/tests/benchmarks/fixtures.py @@ -272,6 +272,76 @@ def enter_component( return enter_component +@dataclass +class Order: + """A row of the table state used by the holistic event benchmark.""" + + name: str + customer: str + amount: float + status: str + + +class TableState(rx.State): + """A state with a 1000-row table, a filter, and derived views of the rows. + + One event on it drives the whole per-event runtime path: a base var + assignment, iterating proxied dataclass rows, sorting them, re-running + the computed vars with their return-type checks, and a delta carrying + hundreds of rows. + """ + + 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_key: rx.Field[str] = rx.field("amount") + 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: getattr(order, self.sort_key), + 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..ffa07dd8e81 100644 --- a/tests/benchmarks/test_event_processing.py +++ b/tests/benchmarks/test_event_processing.py @@ -1,47 +1,61 @@ -"""Benchmark for the event processing pipeline. +"""Benchmarks 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. +Events are enqueued via ``BaseStateEventProcessor`` against a real +``StateManagerMemory`` and every emitted delta is collected. The +``test_process_event`` benchmark times the pipeline alone; the table +benchmark also encodes each delta for the wire the way the socket path does. """ 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 +from .fixtures import BenchmarkState, TableState +RunEvents = Callable[[int, int], Awaitable[None]] -@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. +@asynccontextmanager +async def _event_pipeline( + handler_name: str, + payloads: list[dict[str, Any]], + on_delta: Callable[[Mapping[str, Mapping[str, Any]]], None], +) -> AsyncIterator[RunEvents]: + """Wire a ``BaseStateEventProcessor`` to an in-memory state manager. + + Args: + handler_name: The formatted event handler name to enqueue. + payloads: The payloads to cycle through, one per enqueued event. + 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 enqueues the given number of events and + waits for all expected deltas. """ - 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 @@ -68,45 +82,88 @@ def handle_backend_exception(ex: Exception) -> None: 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={}, - ) + events = [ + Event( + name=handler_name, + router_data={"query": {}, "path": "/"}, + payload=payload, + ) + for payload in payloads + ] 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_events: Number of events to enqueue, cycling the payloads. num_expected_deltas: How many deltas to wait for. """ - emitted_deltas.clear() + 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(token, events[i % len(events)]) + for i in range(num_events) ]): pass - assert len(emitted_deltas) == num_expected_deltas + assert emitted == num_expected_deltas - yield run_events + try: + yield run_events + finally: + await state_manager.close() - await state_manager.close() + +@pytest_asyncio.fixture +async def event_processing_harness(): + """Set up the pipeline for ``BenchmarkState.increment`` with a mock emit. + + Yields: + An async callable that enqueues the given number of events + and waits for all expected deltas. + """ + handler_name = format_event_handler(BenchmarkState.event_handlers["increment"]) + async with _event_pipeline(handler_name, [{}], lambda delta: None) as run: + yield run + + +@pytest_asyncio.fixture +async def table_event_harness(): + """Set up the pipeline for ``TableState.set_status`` with wire encoding. + + Deltas use Reflex's JSON encoder and StateUpdate envelope. Socket.IO + packet framing, network transport, and frontend rendering are excluded. + + Yields: + An async callable that enqueues the given number of events + and waits for all expected deltas. + """ + handler_name = format_event_handler(TableState.event_handlers["set_status"]) + payloads = [{"status": "open"}, {"status": ""}, {"status": "paid"}] + + def encode(delta: Mapping[str, Mapping[str, Any]]) -> None: + """Serialize a delta using the application's JSON encoder. + + Args: + delta: The state changes emitted by the processor. + """ + json_dumps(StateUpdate(delta=delta), separators=(",", ":")) + + async with _event_pipeline(handler_name, payloads, encode) as run: + # Warm caches and finish two filter cycles so sort direction repeats. + await run(6, 6) + yield run def test_process_event( - event_processing_harness, + event_processing_harness: RunEvents, 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. + The first invocation creates fresh state; subsequent invocations reuse + it. Processor startup and shutdown are included in the timing. Args: event_processing_harness: The run_events async callable. @@ -119,4 +176,70 @@ def test_process_event( # 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(run_events(3, 3)) + + +def test_process_table_event( + table_event_harness: RunEvents, + benchmark: BenchmarkFixture, +): + """Benchmark 6 filter events on a warm 1000-row table with JSON encoding. + + Every event assigns base vars, iterates the proxied rows, sorts them, + recomputes both computed vars with their return-type checks, and + produces a delta of hundreds of dataclass rows that is encoded like a + real update. + + Args: + table_event_harness: The run_events async callable. + benchmark: The codspeed benchmark fixture. + """ + run_events = table_event_harness + loop = asyncio.get_event_loop() + + @benchmark + def _(): + loop.run_until_complete(run_events(6, 6)) + + +@pytest.mark.asyncio +async def test_table_event_deltas(): + """Verify filtered rows, sort direction, and totals across repeated batches.""" + updates: list[str] = [] + + def record(delta: Mapping[str, Mapping[str, Any]]) -> None: + """Capture serialized updates for assertions outside the benchmark. + + Args: + delta: The state changes emitted by the processor. + """ + updates.append(json_dumps(StateUpdate(delta=delta), separators=(",", ":"))) + + statuses = ["open", "", "paid"] + handler_name = format_event_handler(TableState.event_handlers["set_status"]) + async with _event_pipeline( + handler_name, [{"status": status} for status in statuses], record + ) as run: + for _ in range(2): + await run(6, 6) + + for index, update in enumerate(updates): + status = statuses[index % len(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 + ) From 8dedc2d1fde615a9c77eae307808445157ceaf9e Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 9 Sep 2026 20:22:42 +0500 Subject: [PATCH 2/2] Simplify table event benchmark setup --- tests/benchmarks/README.md | 24 +-- tests/benchmarks/fixtures.py | 13 +- tests/benchmarks/test_event_processing.py | 213 +++++----------------- 3 files changed, 51 insertions(+), 199 deletions(-) diff --git a/tests/benchmarks/README.md b/tests/benchmarks/README.md index 3656815ebc1..2ec666ec02b 100644 --- a/tests/benchmarks/README.md +++ b/tests/benchmarks/README.md @@ -1,26 +1,16 @@ # Event processing benchmarks -Run the event benchmarks with: - ```sh uv run pytest tests/benchmarks/test_event_processing.py --codspeed ``` Omit `--codspeed` for local wall-clock measurements. -`test_process_event` measures three counter increments through the event processor. -`test_process_table_event` models an order-management dashboard with 1,000 -dataclass rows. Each sample runs six events: filter open orders, clear the filter, -and filter paid orders, twice, reversing the sort direction on every event. -Each update includes sorted matching rows and a computed total. - -The table benchmark warms the state before timing and returns to the same filter -and sort direction after each sample. It covers event queueing, in-memory state -access, mutations, proxied row iteration, computed-variable invalidation and -evaluation, delta generation, and Reflex JSON serialization. Processor startup -and shutdown are included. Initial hydration, Socket.IO packet framing, network -transport, database access, and browser rendering are excluded. +- `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. -`test_table_event_deltas` checks serialized results across two batches separately -from timing, so missing rows or stale computed values cannot silently look like -performance improvements. +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 f417019d962..f5afda90746 100644 --- a/tests/benchmarks/fixtures.py +++ b/tests/benchmarks/fixtures.py @@ -274,7 +274,7 @@ def enter_component( @dataclass class Order: - """A row of the table state used by the holistic event benchmark.""" + """An order in the table event benchmark.""" name: str customer: str @@ -283,13 +283,7 @@ class Order: class TableState(rx.State): - """A state with a 1000-row table, a filter, and derived views of the rows. - - One event on it drives the whole per-event runtime path: a base var - assignment, iterating proxied dataclass rows, sorting them, re-running - the computed vars with their return-type checks, and a delta carrying - hundreds of rows. - """ + """A 1000-row table with filtering, sorting, and a computed total.""" orders: rx.Field[list[Order]] = rx.field( default_factory=lambda: [ @@ -303,7 +297,6 @@ class TableState(rx.State): ] ) status: rx.Field[str] = rx.field("") - sort_key: rx.Field[str] = rx.field("amount") sort_reverse: rx.Field[bool] = rx.field(False) @rx.event @@ -328,7 +321,7 @@ def filtered_orders(self) -> list[Order]: orders = [order for order in orders if order.status == self.status] return sorted( orders, - key=lambda order: getattr(order, self.sort_key), + key=lambda order: order.amount, reverse=self.sort_reverse, ) diff --git a/tests/benchmarks/test_event_processing.py b/tests/benchmarks/test_event_processing.py index ffa07dd8e81..96639a70b79 100644 --- a/tests/benchmarks/test_event_processing.py +++ b/tests/benchmarks/test_event_processing.py @@ -1,23 +1,14 @@ -"""Benchmarks for the event processing pipeline. - -Events are enqueued via ``BaseStateEventProcessor`` against a real -``StateManagerMemory`` and every emitted delta is collected. The -``test_process_event`` benchmark times the pipeline alone; the table -benchmark also encodes each delta for the wire the way the socket path does. -""" +"""Benchmark counter and table events through the in-memory event pipeline.""" import asyncio -import json import traceback -from collections.abc import AsyncIterator, Awaitable, Callable, Mapping -from contextlib import asynccontextmanager +from collections.abc import Mapping 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 @@ -28,26 +19,36 @@ from .fixtures import BenchmarkState, TableState -RunEvents = Callable[[int, int], Awaitable[None]] - -@asynccontextmanager -async def _event_pipeline( - handler_name: str, - payloads: list[dict[str, Any]], - on_delta: Callable[[Mapping[str, Mapping[str, Any]]], None], -) -> AsyncIterator[RunEvents]: - """Wire a ``BaseStateEventProcessor`` to an in-memory state manager. +@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. Args: - handler_name: The formatted event handler name to enqueue. - payloads: The payloads to cycle through, one per enqueued event. - on_delta: Called with each emitted delta. + 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. """ + 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 @@ -55,7 +56,8 @@ async def emit_delta_impl( # noqa: RUF029 ) -> None: nonlocal emitted emitted += 1 - on_delta(delta) + if table: + json_dumps(StateUpdate(delta=delta), separators=(",", ":")) async def emit_event_impl(token: str, *events: Event) -> None: pass @@ -68,178 +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" - events = [ - Event( - name=handler_name, - router_data={"query": {}, "path": "/"}, - payload=payload, - ) - for payload in payloads - ] - - 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 events to enqueue, cycling the payloads. - num_expected_deltas: How many deltas to wait for. - """ + + 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, events[i % len(events)]) - for i in range(num_events) + await p.enqueue("benchmark-token", event) for event in events ]): pass - assert emitted == num_expected_deltas + assert emitted == len(events) try: + if table: + await run_events() yield run_events finally: await state_manager.close() -@pytest_asyncio.fixture -async def event_processing_harness(): - """Set up the pipeline for ``BenchmarkState.increment`` with a mock emit. - - Yields: - An async callable that enqueues the given number of events - and waits for all expected deltas. - """ - handler_name = format_event_handler(BenchmarkState.event_handlers["increment"]) - async with _event_pipeline(handler_name, [{}], lambda delta: None) as run: - yield run - - -@pytest_asyncio.fixture -async def table_event_harness(): - """Set up the pipeline for ``TableState.set_status`` with wire encoding. - - Deltas use Reflex's JSON encoder and StateUpdate envelope. Socket.IO - packet framing, network transport, and frontend rendering are excluded. - - Yields: - An async callable that enqueues the given number of events - and waits for all expected deltas. - """ - handler_name = format_event_handler(TableState.event_handlers["set_status"]) - payloads = [{"status": "open"}, {"status": ""}, {"status": "paid"}] - - def encode(delta: Mapping[str, Mapping[str, Any]]) -> None: - """Serialize a delta using the application's JSON encoder. - - Args: - delta: The state changes emitted by the processor. - """ - json_dumps(StateUpdate(delta=delta), separators=(",", ":")) - - async with _event_pipeline(handler_name, payloads, encode) as run: - # Warm caches and finish two filter cycles so sort direction repeats. - await run(6, 6) - yield run - - -def test_process_event( - event_processing_harness: RunEvents, - benchmark: BenchmarkFixture, -): - """Benchmark processing 3 increment events through the full pipeline. - - The first invocation creates fresh state; subsequent invocations reuse - it. Processor startup and shutdown are included in the timing. +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(3, 3)) - - -def test_process_table_event( - table_event_harness: RunEvents, - benchmark: BenchmarkFixture, -): - """Benchmark 6 filter events on a warm 1000-row table with JSON encoding. - - Every event assigns base vars, iterates the proxied rows, sorts them, - recomputes both computed vars with their return-type checks, and - produces a delta of hundreds of dataclass rows that is encoded like a - real update. - - Args: - table_event_harness: The run_events async callable. - benchmark: The codspeed benchmark fixture. - """ - run_events = table_event_harness - loop = asyncio.get_event_loop() - - @benchmark - def _(): - loop.run_until_complete(run_events(6, 6)) - - -@pytest.mark.asyncio -async def test_table_event_deltas(): - """Verify filtered rows, sort direction, and totals across repeated batches.""" - updates: list[str] = [] - - def record(delta: Mapping[str, Mapping[str, Any]]) -> None: - """Capture serialized updates for assertions outside the benchmark. - - Args: - delta: The state changes emitted by the processor. - """ - updates.append(json_dumps(StateUpdate(delta=delta), separators=(",", ":"))) - - statuses = ["open", "", "paid"] - handler_name = format_event_handler(TableState.event_handlers["set_status"]) - async with _event_pipeline( - handler_name, [{"status": status} for status in statuses], record - ) as run: - for _ in range(2): - await run(6, 6) - - for index, update in enumerate(updates): - status = statuses[index % len(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 - ) + loop.run_until_complete(event_processing_harness())