Expand process benchmarks across lifecycle phases and topologies

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

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

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.

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())

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

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()