From d2f681a69479d5c169bdcc67dd7da17464248a50 Mon Sep 17 00:00:00 2001 From: Farhan Date: Mon, 7 Sep 2026 22:55:24 +0500 Subject: [PATCH 1/3] 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/3] 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()) From fca0ca544475b6ebd29bc814208a0278a6f663b9 Mon Sep 17 00:00:00 2001 From: Farhan Date: Wed, 16 Sep 2026 21:32:42 +0000 Subject: [PATCH 3/3] Restore table delta correctness test and keep batches net-zero Re-add test_table_event_deltas, dropped in the harness simplification, so the serialized rows, sort direction, and totals the table benchmark encodes are verified across repeated batches, sharing one _encode_delta helper with the timed path. Replace the counter batch's three increments with two increments and two decrements: the old batch grew the counter every sample, so the computed vars derived from it did strictly more work on each invocation. Both workloads now return state to its starting point between samples. --- tests/benchmarks/README.md | 10 +- tests/benchmarks/test_event_processing.py | 145 ++++++++++++++++++---- 2 files changed, 127 insertions(+), 28 deletions(-) diff --git a/tests/benchmarks/README.md b/tests/benchmarks/README.md index 2ec666ec02b..379ca638fd5 100644 --- a/tests/benchmarks/README.md +++ b/tests/benchmarks/README.md @@ -6,11 +6,15 @@ 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. +- `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; each batch returns to the same filter and -sort direction. Timing includes processor startup/shutdown, but excludes initial +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/test_event_processing.py b/tests/benchmarks/test_event_processing.py index 96639a70b79..c4fb3e13d95 100644 --- a/tests/benchmarks/test_event_processing.py +++ b/tests/benchmarks/test_event_processing.py @@ -1,14 +1,17 @@ """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 @@ -19,36 +22,76 @@ from .fixtures import BenchmarkState, TableState +TABLE_STATUSES = ("open", "", "paid") -@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 _encode_delta(delta: Mapping[str, Mapping[str, Any]]) -> str: + """Serialize a delta the way the socket path does. Args: - request: Selects the counter or table workload. + delta: The state changes emitted by the processor. - Yields: - An async callable that processes one batch and checks its delta count. + Returns: + The delta encoded as a StateUpdate envelope. """ - 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 = [ + return json_dumps(StateUpdate(delta=delta), separators=(",", ":")) + + +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=format_event_handler(handler), + name=handler_name, router_data={"query": {}, "path": "/"}, payload=payload, ) for payload in payloads ] + + +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 processes one batch and checks its delta count. + """ emitted = 0 async def emit_delta_impl( # noqa: RUF029 @@ -56,8 +99,7 @@ async def emit_delta_impl( # noqa: RUF029 ) -> None: nonlocal emitted emitted += 1 - if table: - json_dumps(StateUpdate(delta=delta), separators=(",", ":")) + on_delta(delta) async def emit_event_impl(token: str, *events: Event) -> None: pass @@ -93,15 +135,35 @@ async def run_events() -> None: assert emitted == len(events) try: - if table: - await run_events() yield run_events finally: 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. + + Both batches return state to its starting point, so every benchmark + sample measures identical work. + + 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 three counter events or six table events. + """Benchmark a batch of four counter events or six table events. Args: event_processing_harness: The async batch runner. @@ -112,3 +174,36 @@ def test_process_event(event_processing_harness, benchmark: BenchmarkFixture): @benchmark def _(): 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 + )