From e629be287bf589ace5274cba8d4405a2f5300a56 Mon Sep 17 00:00:00 2001 From: Toby Coleman Date: Tue, 20 Oct 2026 21:03:30 +0100 Subject: [PATCH 1/2] fix: receive NumPy arrays without comparing them to channel close marker --- plugboard/connector/channel.py | 2 +- plugboard/connector/serde_channel.py | 2 +- tests/unit/test_channel_numpy.py | 72 ++++++++++++++++++++++++++++ 3 files changed, 74 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_channel_numpy.py diff --git a/plugboard/connector/channel.py b/plugboard/connector/channel.py index 4279e7c2..55a9489b 100644 --- a/plugboard/connector/channel.py +++ b/plugboard/connector/channel.py @@ -85,7 +85,7 @@ async def _wrapper() -> _t.Any: if self._is_recv_closed: raise ChannelClosedError("Attempted recv on closed channel.") msg = await self._recv() - if msg == CHAN_CLOSE_MSG: + if isinstance(msg, str) and msg == CHAN_CLOSE_MSG: await self.close() self._is_recv_closed = True self._is_send_closed = True diff --git a/plugboard/connector/serde_channel.py b/plugboard/connector/serde_channel.py index cd99edd0..b9496859 100644 --- a/plugboard/connector/serde_channel.py +++ b/plugboard/connector/serde_channel.py @@ -64,7 +64,7 @@ async def _wrapper() -> _t.Any: if self._is_recv_closed: raise ChannelClosedError("Attempted recv on closed channel.") msg = _deserialise(await self._recv()) - if msg == CHAN_CLOSE_MSG: + if isinstance(msg, str) and msg == CHAN_CLOSE_MSG: await self.close() self._is_recv_closed = True self._is_send_closed = True diff --git a/tests/unit/test_channel_numpy.py b/tests/unit/test_channel_numpy.py new file mode 100644 index 00000000..e61bea59 --- /dev/null +++ b/tests/unit/test_channel_numpy.py @@ -0,0 +1,72 @@ +"""Regression tests for NumPy payloads in channel receive wrappers.""" + +import asyncio +import typing as _t + +import numpy as np +import numpy.typing as npt +import pytest + +from plugboard.connector.asyncio_channel import AsyncioChannel +from plugboard.connector.channel import CHAN_CLOSE_MSG, Channel +from plugboard.connector.serde_channel import SerdeChannel +from plugboard.exceptions import ChannelClosedError + + +class _InMemorySerdeChannel(SerdeChannel): + """Exercise the serialized channel wrapper without an external broker.""" + + def __init__(self) -> None: + super().__init__() + self._queue: asyncio.Queue[bytes] = asyncio.Queue() + + async def send(self, msg: bytes) -> None: + """Queue a serialized message.""" + await self._queue.put(msg) + + async def recv(self) -> bytes: + """Return a queued serialized message.""" + return await self._queue.get() + + +@pytest.mark.parametrize("channel_cls", [AsyncioChannel, _InMemorySerdeChannel]) +@pytest.mark.parametrize( + "payload", + [ + pytest.param(np.arange(3), id="vector"), + pytest.param(np.arange(6).reshape(2, 3), id="matrix"), + pytest.param(np.array([]), id="empty"), + pytest.param(np.array(42), id="scalar"), + pytest.param(np.array([42]), id="singleton"), + pytest.param(np.array(["a", "b"]), id="strings"), + pytest.param(np.array([1, "a"], dtype=object), id="objects"), + pytest.param(np.array([(1, 2.0)], dtype=[("x", "i4"), ("y", "f8")]), id="structured"), + pytest.param(np.arange(8)[::2], id="noncontiguous"), + pytest.param(np.array(CHAN_CLOSE_MSG), id="close-marker-scalar"), + pytest.param(np.array([CHAN_CLOSE_MSG]), id="close-marker-singleton"), + ], +) +async def test_numpy_payload_round_trip( + channel_cls: type[Channel], payload: npt.NDArray[_t.Any] +) -> None: + """Arrays remain data, and the actual close marker still closes the channel.""" + channel = channel_cls() + await channel.send(payload) + received = await channel.recv() + + assert isinstance(received, np.ndarray) + np.testing.assert_array_equal(received, payload) + assert received.dtype == payload.dtype + assert received.shape == payload.shape + assert not channel.is_closed + + await channel.send("next message") + assert await channel.recv() == "next message" + + await channel.close() + with pytest.raises(ChannelClosedError): + await channel.recv() + with pytest.raises(ChannelClosedError): + await channel.recv() + with pytest.raises(ChannelClosedError): + await channel.send(payload) From 0d7878140996b92637ed086f8a1f1af32e508595 Mon Sep 17 00:00:00 2001 From: Toby Coleman Date: Sat, 24 Oct 2026 22:06:07 +0100 Subject: [PATCH 2/2] Expand process benchmarks across lifecycle phases and topologies --- tests/benchmark/README.md | 53 ++++++++++ tests/benchmark/test_benchmarking.py | 104 ++++++++++++++----- tests/benchmark/workloads.py | 136 +++++++++++++++++++++++++ tests/unit/test_benchmark_lifecycle.py | 57 +++++++++++ 4 files changed, 326 insertions(+), 24 deletions(-) create mode 100644 tests/benchmark/README.md create mode 100644 tests/benchmark/workloads.py create mode 100644 tests/unit/test_benchmark_lifecycle.py diff --git a/tests/benchmark/README.md b/tests/benchmark/README.md new file mode 100644 index 00000000..e8e730b9 --- /dev/null +++ b/tests/benchmark/README.md @@ -0,0 +1,53 @@ +# Process benchmarks + +Run the suite from the repository root: + +```sh +uv run pytest tests/benchmark --benchmark-only --benchmark-save=process +``` + +Use `--benchmark-disable` to execute each case once, including correctness checks +and cleanup. Filter cases with `-k`, for example `-k 'run and fan_in-16'`. +The existing benchmark workflow runs this suite on both the PR and main branch. +New case IDs have no historical baseline until they are present on both branches. + +Each workload uses `LocalProcess`, `AsyncioConnector`, the default dictionary state +backend, and the production `Component.run` loop. Components do minimal integer +work so timings emphasize framework scheduling, channel operations, producer +status checks, and state updates rather than application computation or external +services. Each source emits 1,000 integers. + +| Graph | Scale | Cost exercised | +| --- | --- | --- | +| Chain | 3 and 16 total components | Pipeline depth and per-hop overhead | +| Fan-out | 2 and 16 receivers | Output broadcast and concurrent consumers | +| Fan-in | 2 and 16 sources sharing one input field | Input multiplexing and producer status tracking | +| Diamond | 4 components | Branching followed by synchronization of two input fields | +| Feedback | 3 components, one seeded input | Cyclic dependencies and repeated synchronization | + +Fan-in delivers `sources * 1,000` messages; fan-out delivers 1,000 messages to +each receiver. Increasing width therefore increases total work. The diamond join +waits for both branches and forwards the first value. The feedback join waits for +a source value and the preceding feedback value, with an initial seed to let it +start. Receiver counts and checksums and all component completion statuses are +checked outside the timed region so lost messages cannot look like a speedup. + +Every case reports three timing groups, each with five fresh process instances: + +- `init`: `process.init()` only, including wiring, state connection, component + initialization, and graph validation. Destruction happens outside the timer. +- `run`: `process.run()` only, with initialization and destruction outside the timer. +- `lifecycle`: initialization, execution, and destruction together. + +Graph construction is outside all three timers. Each timed call includes one +`Runner.run()` dispatch; event-loop creation and closure are outside the timer. +Construction, initialization, execution, and destruction all use the same running +event loop for each process. Processes are destroyed even when initialization or +execution fails. No initialization delays are patched: in particular, +`LocalProcess` currently waits one second for connections during initialization, +which dominates `init` and `lifecycle` results for small graphs. Keep the separate +`run` group when examining throughput regressions. + +These are local in-memory benchmarks. They do not measure serialization, +distributed workers, external brokers, or persistent state backends, whose service +startup and environmental variability require separate benchmark infrastructure. diff --git a/tests/benchmark/test_benchmarking.py b/tests/benchmark/test_benchmarking.py index 7554a7a0..f22163cb 100644 --- a/tests/benchmark/test_benchmarking.py +++ b/tests/benchmark/test_benchmarking.py @@ -1,35 +1,91 @@ -"""Simple benchmark tests for Plugboard models.""" +"""Measure lifecycle and execution costs across representative local process graphs.""" import asyncio +import typing as _t +import pytest from pytest_benchmark.fixture import BenchmarkFixture -from plugboard.connector import AsyncioConnector -from plugboard.process import LocalProcess, Process -from plugboard.schemas import ConnectorSpec -from tests.integration.test_process_with_components_run import A, B +from plugboard.process import LocalProcess +from tests.benchmark.workloads import Relay, Workload, verify -def _setup_process() -> tuple[tuple[Process], dict]: - comp_a = A(name="comp_a", iters=1000) - comp_b1 = B(name="comp_b1", factor=1) - comp_b2 = B(name="comp_b2", factor=2) - components = [comp_a, comp_b1, comp_b2] - connectors = [ - AsyncioConnector(spec=ConnectorSpec(source="comp_a.out_1", target="comp_b1.in_1")), - AsyncioConnector(spec=ConnectorSpec(source="comp_b1.out_1", target="comp_b2.in_1")), - ] - process = LocalProcess(components=components, connectors=connectors) - # Initialise process so that this is excluded from the benchmark timing - asyncio.run(process.init()) - # Return args and kwargs tuple for benchmark.pedantic - return (process,), {} +WORKLOADS = [ + Workload("chain", 3), + Workload("chain", 16), + Workload("fan_out", 2), + Workload("fan_out", 16), + Workload("fan_in", 2), + Workload("fan_in", 16), + Workload("diamond", 4), + Workload("feedback", 3), +] +MESSAGES = 1000 +Phase = _t.Literal["init", "run", "lifecycle"] -def _run_process(process: Process) -> None: - asyncio.run(process.run()) +@pytest.mark.parametrize("workload", WORKLOADS, ids=lambda workload: workload.id) +@pytest.mark.parametrize("phase", ["init", "run", "lifecycle"]) +def test_benchmark_process(benchmark: BenchmarkFixture, workload: Workload, phase: Phase) -> None: + """Time one lifecycle phase per fresh graph, using one loop for its entire lifetime.""" + process: LocalProcess | None = None + expected: list[tuple[Relay, int]] = [] + needs_destroy = False + benchmark.group = phase + benchmark.extra_info.update( + topology=workload.topology, size=workload.size, messages_per_source=MESSAGES + ) + with asyncio.Runner() as runner: -def test_benchmark_process_run(benchmark: BenchmarkFixture) -> None: - """Benchmark the running of a Plugboard Process.""" - benchmark.pedantic(_run_process, setup=_setup_process, rounds=5) + async def setup_round() -> None: + nonlocal process, expected, needs_destroy + process, expected = workload.build(MESSAGES) + needs_destroy = True + if phase == "run": + await process.init() + + async def measure() -> None: + nonlocal needs_destroy + assert process is not None + if phase == "lifecycle": + try: + await process.init() + await process.run() + finally: + await process.destroy() + needs_destroy = False + elif phase == "init": + await process.init() + else: + await process.run() + + async def finish_round() -> None: + nonlocal needs_destroy + if process is None: + return + try: + if phase == "init": + assert process.is_initialised + else: + verify(process, expected, MESSAGES) + finally: + if needs_destroy: + await process.destroy() + needs_destroy = False + + try: + benchmark.pedantic( + lambda: runner.run(measure()), + setup=lambda: runner.run(setup_round()), + teardown=lambda: runner.run(finish_round()), + rounds=5, + iterations=1, + ) + # Disabled benchmarks execute once without invoking their teardown callback. + if benchmark.disabled: + runner.run(finish_round()) + finally: + # Failed setup/targets also bypass pytest-benchmark's teardown callback. + if process is not None and needs_destroy: + runner.run(process.destroy()) diff --git a/tests/benchmark/workloads.py b/tests/benchmark/workloads.py new file mode 100644 index 00000000..1129a43c --- /dev/null +++ b/tests/benchmark/workloads.py @@ -0,0 +1,136 @@ +"""Deterministic workloads using the production component run loop.""" + +from dataclasses import dataclass +import typing as _t + +from plugboard.component import Component, IOController as IO +from plugboard.connector import AsyncioConnector +from plugboard.process import LocalProcess +from plugboard.schemas import ComponentArgsDict, ConnectorSpec, Status + + +class Source(Component): + """Emit a finite integer sequence without application computation or I/O.""" + + io = IO(outputs=["value"]) + + def __init__(self, messages: int, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: + super().__init__(**kwargs) + self.messages = messages + self.sent = 0 + + async def step(self) -> None: + """Emit the next integer, closing the stream when exhausted.""" + if self.sent == self.messages: + await self.io.close() + else: + self.value = self.sent + self.sent += 1 + + +class Relay(Component): + """Forward values while recording delivery counts and checksums.""" + + io = IO(inputs=["value"], outputs=["result"]) + + def __init__(self, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: + super().__init__(**kwargs) + self.received = 0 + self.checksum = 0 + + async def step(self) -> None: + """Forward one input through the normal component I/O wrapper.""" + self.result = self.value + self.received += 1 + self.checksum += self.value + + +class Join(Relay): + """Synchronize two inputs, forwarding the first after both arrive.""" + + io = IO(inputs=["value", "other"], outputs=["result"]) + + +@dataclass(frozen=True) +class Workload: + """A graph shape and its scale, with stable benchmark identifiers.""" + + topology: str + size: int + + @property + def id(self) -> str: + """Return the topology and scale for pytest result names.""" + return f"{self.topology}-{self.size}" + + def build(self, messages: int) -> tuple[LocalProcess, list[tuple[Relay, int]]]: + """Construct a fresh process and expected message counts per receiver.""" + components: list[Component] = [] + connectors: list[AsyncioConnector] = [] + expected: list[tuple[Relay, int]] = [] + + def source(name: str) -> Source: + component = Source(messages=messages, name=name) + components.append(component) + return component + + def relay(name: str, count: int = messages, *, join: bool = False) -> Relay: + component = Join(name=name) if join else Relay(name=name) + components.append(component) + expected.append((component, count)) + return component + + def connect(sender: Component, receiver: Component, field: str = "value") -> None: + output = "value" if isinstance(sender, Source) else "result" + connectors.append( + AsyncioConnector( + spec=ConnectorSpec( + source=f"{sender.name}.{output}", target=f"{receiver.name}.{field}" + ) + ) + ) + + if self.topology == "chain": + previous: Component = source("source") + for index in range(self.size - 1): + receiver = relay(f"relay_{index}") + connect(previous, receiver) + previous = receiver + elif self.topology == "fan_out": + producer = source("source") + for index in range(self.size): + connect(producer, relay(f"receiver_{index}")) + elif self.topology == "fan_in": + receiver = relay("receiver", messages * self.size) + for index in range(self.size): + connect(source(f"source_{index}"), receiver) + elif self.topology == "diamond": + producer = source("source") + left, right = relay("left"), relay("right") + receiver = relay("join", join=True) + connect(producer, left) + connect(producer, right) + connect(left, receiver) + connect(right, receiver, "other") + elif self.topology == "feedback": + producer = source("source") + receiver = Join(name="join", initial_values={"other": [-1]}) + components.append(receiver) + expected.append((receiver, messages)) + feedback = relay("feedback") + connect(producer, receiver) + connect(receiver, feedback) + connect(feedback, receiver, "other") + else: + raise ValueError(f"Unknown topology: {self.topology}") + + return LocalProcess(components, connectors, name=self.id), expected + + +def verify(process: LocalProcess, expected: list[tuple[Relay, int]], messages: int) -> None: + """Reject incomplete runs even if they appear faster.""" + assert process.status == Status.COMPLETED + assert all(component.status == Status.COMPLETED for component in process.components.values()) + for component, count in expected: + assert component.received == count + assert component.checksum == count * (messages - 1) // 2 diff --git a/tests/unit/test_benchmark_lifecycle.py b/tests/unit/test_benchmark_lifecycle.py new file mode 100644 index 00000000..52850ea9 --- /dev/null +++ b/tests/unit/test_benchmark_lifecycle.py @@ -0,0 +1,57 @@ +"""Regression checks for benchmark event-loop ownership and failure cleanup.""" + +import asyncio +from collections.abc import Callable +from unittest.mock import Mock + +import pytest +from pytest_benchmark.fixture import BenchmarkFixture + +from tests.benchmark.test_benchmarking import Phase, test_benchmark_process as run_benchmark +from tests.benchmark.workloads import Workload + + +@pytest.mark.parametrize("phase", ["init", "run", "lifecycle"]) +def test_benchmark_destroys_failed_process_on_its_own_loop(phase: Phase) -> None: + """Failed initialization/execution must clean up before the owning loop closes.""" + loops: list[asyncio.AbstractEventLoop] = [] + + class FailingProcess: + async def init(self) -> None: + loops.append(asyncio.get_running_loop()) + if phase != "run": + raise RuntimeError("initialization failed") + + async def run(self) -> None: + loops.append(asyncio.get_running_loop()) + raise RuntimeError("execution failed") + + async def destroy(self) -> None: + loops.append(asyncio.get_running_loop()) + + workload = Mock(spec=Workload) + workload.topology = "chain" + workload.size = 3 + workload.build.return_value = (FailingProcess(), []) + benchmark = Mock(spec=BenchmarkFixture) + benchmark.extra_info = {} + + def pedantic( + target: Callable[[], None], + *, + setup: Callable[[], None], + teardown: Callable[[], None], + rounds: int, + iterations: int, + ) -> None: + # Like pytest-benchmark, failure skips the teardown callback. + setup() + target() + + benchmark.pedantic.side_effect = pedantic + with pytest.raises(RuntimeError, match="failed"): + run_benchmark(benchmark, workload, phase) + + assert len(loops) == (3 if phase == "run" else 2) + assert all(loop is loops[0] for loop in loops) + assert loops[0].is_closed()