From 9ac8db27e9e1233b091a4ed7341d25fa50588649 Mon Sep 17 00:00:00 2001 From: Tim Paine <3105306+timkpaine@users.noreply.github.com> Date: Sat, 1 Aug 2026 13:52:12 -0400 Subject: [PATCH] Add stream-contract conformance checkers Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com> --- dau_sim/integrations/protocol.py | 291 +++++++++++++++++++++ dau_sim/tests/test_protocol_conformance.py | 146 +++++++++++ docs/src/cocotb.md | 18 ++ 3 files changed, 455 insertions(+) create mode 100644 dau_sim/integrations/protocol.py create mode 100644 dau_sim/tests/test_protocol_conformance.py diff --git a/dau_sim/integrations/protocol.py b/dau_sim/integrations/protocol.py new file mode 100644 index 0000000..2800608 --- /dev/null +++ b/dau_sim/integrations/protocol.py @@ -0,0 +1,291 @@ +"""Runtime checkers for DAU valid/ready stream contracts.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Literal + + +class StreamContractViolation(AssertionError): + """A sampled interface violated its stream contract.""" + + +class HandshakeContractChecker: + """Simulator-neutral valid/ready stability checker.""" + + def __init__( + self, + name: str, + payload_names: Sequence[str], + *, + valid_rule: str = "VALID_HELD_UNTIL_READY", + payload_rule: str = "PAYLOAD_STABLE_UNTIL_READY", + ) -> None: + if not name: + raise ValueError("interface name must be non-empty") + if not payload_names: + raise ValueError("at least one payload signal is required") + self.name = name + self.payload_names = tuple(payload_names) + self._valid_rule = valid_rule + self._payload_rule = payload_rule + self.cycle = 0 + self.transfers = 0 + self._held_payload: tuple[int, ...] | None = None + + def reset(self) -> None: + self.cycle = 0 + self.transfers = 0 + self._held_payload = None + + def observe(self, *, valid: int, ready: int, payload: Mapping[str, int]) -> bool: + """Sample one active clock edge and return whether a transfer fired.""" + self.cycle += 1 + valid_bit = self._binary_value("valid", valid) + ready_bit = self._binary_value("ready", ready) + values = tuple(int(payload[name]) for name in self.payload_names) + + if self._held_payload is not None: + if not valid_bit: + self._violate(self._valid_rule, "valid dropped before ready accepted the stalled transfer") + if values != self._held_payload: + changed = [ + name for name, previous, current in zip(self.payload_names, self._held_payload, values, strict=True) if previous != current + ] + self._violate(self._payload_rule, f"stalled payload changed: {', '.join(changed)}") + + transfer = bool(valid_bit and ready_bit) + if transfer: + self.transfers += 1 + self._held_payload = None + elif valid_bit: + self._held_payload = values + else: + self._held_payload = None + return transfer + + def finish(self) -> None: + """Validate end-of-observation rules.""" + + def _binary_value(self, signal: str, value: int) -> int: + bit = int(value) + if bit not in (0, 1): + self._violate("BINARY_CONTROL", f"{signal} must be 0 or 1, got {bit}") + return bit + + def _violate(self, rule: str, detail: str) -> None: + raise StreamContractViolation(f"[{rule}] {self.name} cycle {self.cycle}: {detail}") + + +class StreamContractChecker(HandshakeContractChecker): + """Check valid/ready stability and optional ``last`` framing.""" + + def __init__( + self, + name: str, + *, + payload_names: Sequence[str] = ("data", "last"), + last_name: str | None = "last", + expected_batches: int | None = None, + ) -> None: + if expected_batches is not None and expected_batches < 0: + raise ValueError("expected_batches must be non-negative") + if last_name is not None and last_name not in payload_names: + raise ValueError("last_name must be included in payload_names") + super().__init__(name, payload_names) + self.last_name = last_name + self.expected_batches = expected_batches + self.completed_batches = 0 + + def reset(self) -> None: + super().reset() + self.completed_batches = 0 + + def observe(self, *, valid: int, ready: int, payload: Mapping[str, int]) -> bool: + transfer = super().observe(valid=valid, ready=ready, payload=payload) + if not transfer or self.last_name is None: + return transfer + + last = self._binary_value(self.last_name, payload[self.last_name]) + if self.expected_batches is not None and self.completed_batches >= self.expected_batches: + if last: + self._violate("LAST_EXACTLY_ONCE_PER_BATCH", "duplicate last transfer after all expected batches completed") + self._violate("NO_TRANSFER_AFTER_LAST", "transfer occurred after the final batch's last") + if last: + self.completed_batches += 1 + return transfer + + def finish(self) -> None: + if self.expected_batches is not None and self.completed_batches != self.expected_batches: + self._violate( + "LAST_EXACTLY_ONCE_PER_BATCH", + f"missing last transfer: completed {self.completed_batches} of {self.expected_batches} expected batches", + ) + + +StatusMode = Literal["terminal", "mid_lane"] + + +class StatusContractChecker(HandshakeContractChecker): + """Check status backpressure and terminal or mid-lane cardinality.""" + + def __init__( + self, + name: str = "status_", + *, + mode: StatusMode, + expected_batches: int | None = None, + payload_names: Sequence[str] = ("error", "error_code"), + error_name: str = "error", + ) -> None: + if mode not in ("terminal", "mid_lane"): + raise ValueError("mode must be 'terminal' or 'mid_lane'") + if expected_batches is not None and expected_batches < 0: + raise ValueError("expected_batches must be non-negative") + if error_name not in payload_names: + raise ValueError("error_name must be included in payload_names") + super().__init__( + name, + payload_names, + valid_rule="STATUS_HELD_UNTIL_READY", + payload_rule="STATUS_PAYLOAD_STABLE_UNTIL_READY", + ) + self.mode = mode + self.expected_batches = expected_batches + self.error_name = error_name + self.completed_statuses = 0 + + def reset(self) -> None: + super().reset() + self.completed_statuses = 0 + + def observe(self, *, valid: int, ready: int, payload: Mapping[str, int]) -> bool: + transfer = super().observe(valid=valid, ready=ready, payload=payload) + if not transfer: + return False + + error = self._binary_value(self.error_name, payload[self.error_name]) + if self.mode == "mid_lane" and not error: + self._violate("MID_LANE_STATUS_ONLY_ON_ERROR", "mid-lane emitted a success status") + self.completed_statuses += 1 + if self.expected_batches is not None and self.completed_statuses > self.expected_batches: + self._violate("STATUS_EXACTLY_ONCE_PER_BATCH", "more statuses than expected batches") + return True + + def finish(self) -> None: + if self.mode == "terminal" and self.expected_batches is not None and self.completed_statuses != self.expected_batches: + self._violate( + "STATUS_EXACTLY_ONCE_PER_BATCH", + f"completed {self.completed_statuses} statuses for {self.expected_batches} expected batches", + ) + + +class _CocotbContractMonitor: + def __init__(self, dut, clock, prefix: str, checker: HandshakeContractChecker, *, reset=None) -> None: + self.checker = checker + self.clock = clock + self.reset = reset + self._valid = self._resolve(dut, f"{prefix}valid") + self._ready = self._resolve(dut, f"{prefix}ready") + self._payload = {name: self._resolve(dut, f"{prefix}{name}") for name in checker.payload_names} + self._task = None + self._violation: StreamContractViolation | None = None + + async def __aenter__(self): + import cocotb + + self._task = cocotb.start_soon(self._run()) + return self + + async def __aexit__(self, exc_type, exc_value, traceback) -> bool: + if self._task is not None and not self._task.done(): + self._task.kill() + if exc_type is not None: + return False + if self._violation is not None: + raise self._violation + self.checker.finish() + return False + + async def _run(self) -> None: + from cocotb.triggers import RisingEdge + + while True: + await RisingEdge(self.clock) + if self.reset is not None and int(self.reset.value): + self.checker.reset() + continue + try: + self.checker.observe( + valid=int(self._valid.value), + ready=int(self._ready.value), + payload={name: int(handle.value) for name, handle in self._payload.items()}, + ) + except StreamContractViolation as exc: + self._violation = exc + return + + @staticmethod + def _resolve(dut, name: str): + try: + return getattr(dut, name) + except AttributeError as exc: + raise ValueError(f"DUT has no contract signal {name!r}") from exc + + +class StreamContractMonitor(_CocotbContractMonitor): + """Cocotb context manager attaching a checker to a prefixed stream.""" + + def __init__( + self, + dut, + clock, + prefix: str, + *, + reset=None, + payload_names: Sequence[str] = ("data", "last"), + last_name: str | None = "last", + expected_batches: int | None = None, + ) -> None: + checker = StreamContractChecker( + prefix, + payload_names=payload_names, + last_name=last_name, + expected_batches=expected_batches, + ) + super().__init__(dut, clock, prefix, checker, reset=reset) + + +class StatusContractMonitor(_CocotbContractMonitor): + """Cocotb context manager attaching a checker to a status interface.""" + + def __init__( + self, + dut, + clock, + *, + prefix: str = "status_", + reset=None, + mode: StatusMode, + expected_batches: int | None = None, + payload_names: Sequence[str] = ("error", "error_code"), + error_name: str = "error", + ) -> None: + checker = StatusContractChecker( + prefix, + mode=mode, + expected_batches=expected_batches, + payload_names=payload_names, + error_name=error_name, + ) + super().__init__(dut, clock, prefix, checker, reset=reset) + + +__all__ = ( + "HandshakeContractChecker", + "StatusContractChecker", + "StatusContractMonitor", + "StreamContractChecker", + "StreamContractMonitor", + "StreamContractViolation", +) diff --git a/dau_sim/tests/test_protocol_conformance.py b/dau_sim/tests/test_protocol_conformance.py new file mode 100644 index 0000000..3eef924 --- /dev/null +++ b/dau_sim/tests/test_protocol_conformance.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import pytest + +from dau_sim.compiler import compile_module +from dau_sim.integrations.protocol import StatusContractChecker, StreamContractChecker, StreamContractMonitor, StreamContractViolation +from dau_sim.ir import Assign, Binary, BinaryOp, ClockDomain, Const, EdgePolarity, Module, Port, PortDirection, SeqBlock, Shape, Signal, SignalRef +from dau_sim.tests.test_cocotb_examples import CocotbExampleTestBase + + +def _broken_stream_module(failure: str) -> Module: + count = SignalRef(shape=Shape(4), name="count") + valid = Const(shape=Shape(1), value=1) + data = Const(shape=Shape(8), value=17) + last = Const(shape=Shape(1), value=0) + + if failure == "drops_valid": + valid = Binary(shape=Shape(1), op=BinaryOp.EQ, left=count, right=Const(shape=Shape(4), value=0)) + elif failure == "mutates_payload": + data = Binary(shape=Shape(8), op=BinaryOp.ADD, left=SignalRef(shape=Shape(8), name="output_data"), right=Const(shape=Shape(8), value=1)) + elif failure == "emits_two_lasts": + last = Const(shape=Shape(1), value=1) + elif failure != "never_emits_last": + raise ValueError(f"unknown failure {failure!r}") + + return Module( + name=f"broken_stream_{failure}", + ports=( + Port(Signal("clk", Shape(1)), PortDirection.INPUT), + Port(Signal("output_ready", Shape(1)), PortDirection.INPUT), + Port(Signal("output_valid", Shape(1)), PortDirection.OUTPUT), + Port(Signal("output_data", Shape(8)), PortDirection.OUTPUT), + Port(Signal("output_last", Shape(1)), PortDirection.OUTPUT), + ), + signals=(Signal("count", Shape(4)),), + clock_domains=(ClockDomain("sync", clk="clk", edge=EdgePolarity.POSEDGE),), + seq_blocks=( + SeqBlock( + domain="sync", + stmts=( + Assign("output_valid", valid), + Assign("output_data", data), + Assign("output_last", last), + Assign("count", Binary(shape=Shape(4), op=BinaryOp.ADD, left=count, right=Const(shape=Shape(4), value=1))), + ), + ), + ), + ) + + +def _check_broken_module(failure: str, *, ready: int) -> None: + module = _broken_stream_module(failure) + traces = compile_module(module).run(cycles=4, inputs={"output_ready": ready}) + checker = StreamContractChecker("output_", expected_batches=1) + for index in range(len(traces["clk"])): + checker.observe( + valid=traces["output_valid"][index][1], + ready=traces["output_ready"][index][1], + payload={"data": traces["output_data"][index][1], "last": traces["output_last"][index][1]}, + ) + checker.finish() + + +def test_broken_module_that_drops_valid_identifies_hold_rule() -> None: + with pytest.raises(StreamContractViolation, match=r"\[VALID_HELD_UNTIL_READY\] output_ cycle 2"): + _check_broken_module("drops_valid", ready=0) + + +def test_broken_module_that_mutates_payload_identifies_stability_rule() -> None: + with pytest.raises(StreamContractViolation, match=r"\[PAYLOAD_STABLE_UNTIL_READY\] output_ cycle 2"): + _check_broken_module("mutates_payload", ready=0) + + +def test_broken_module_that_emits_two_lasts_identifies_last_rule() -> None: + with pytest.raises(StreamContractViolation, match=r"\[LAST_EXACTLY_ONCE_PER_BATCH\] output_ cycle 2"): + _check_broken_module("emits_two_lasts", ready=1) + + +def test_broken_module_that_never_emits_last_identifies_last_rule() -> None: + with pytest.raises(StreamContractViolation, match=r"\[LAST_EXACTLY_ONCE_PER_BATCH\] output_ cycle 4"): + _check_broken_module("never_emits_last", ready=1) + + +def test_transfer_count_only_advances_when_valid_and_ready_are_high() -> None: + checker = StreamContractChecker("input_", expected_batches=None) + for valid, ready in ((0, 0), (0, 1), (1, 0), (1, 1)): + checker.observe(valid=valid, ready=ready, payload={"data": 9, "last": 0}) + assert checker.transfers == 1 + + +def test_transfer_after_final_last_is_rejected() -> None: + checker = StreamContractChecker("output_", expected_batches=1) + checker.observe(valid=1, ready=1, payload={"data": 9, "last": 1}) + with pytest.raises(StreamContractViolation, match="NO_TRANSFER_AFTER_LAST"): + checker.observe(valid=1, ready=1, payload={"data": 10, "last": 0}) + + +def test_terminal_status_requires_one_held_status_per_expected_batch() -> None: + checker = StatusContractChecker(mode="terminal", expected_batches=1) + checker.observe(valid=1, ready=0, payload={"error": 0, "error_code": 0}) + checker.observe(valid=1, ready=1, payload={"error": 0, "error_code": 0}) + checker.finish() + assert checker.completed_statuses == 1 + + +def test_status_must_be_held_until_ready() -> None: + checker = StatusContractChecker(mode="terminal", expected_batches=1) + checker.observe(valid=1, ready=0, payload={"error": 0, "error_code": 0}) + with pytest.raises(StreamContractViolation, match="STATUS_HELD_UNTIL_READY"): + checker.observe(valid=0, ready=0, payload={"error": 0, "error_code": 0}) + + +def test_mid_lane_rejects_success_status() -> None: + checker = StatusContractChecker(mode="mid_lane", expected_batches=1) + with pytest.raises(StreamContractViolation, match="MID_LANE_STATUS_ONLY_ON_ERROR"): + checker.observe(valid=1, ready=1, payload={"error": 0, "error_code": 0}) + + +class TestCocotbStreamContractMonitor(CocotbExampleTestBase): + def test_context_manager_attaches_to_prefixed_interface(self) -> None: + import cocotb + + engine = self._setup(_broken_stream_module("mutates_payload")) + violations: list[str] = [] + try: + from cocotb._gpi_triggers import RisingEdge + from cocotb.clock import Clock + + async def run_bench() -> None: + dut = cocotb.top + dut.output_ready.value = 0 + Clock(dut.clk, 10, unit="ns").start(start_high=False) + try: + async with StreamContractMonitor(dut, dut.clk, "output_", expected_batches=1): + for _ in range(4): + await RisingEdge(dut.clk) + except StreamContractViolation as exc: + violations.append(str(exc)) + engine.stop() + + self._run_coroutine(run_bench()) + finally: + self._teardown() + + assert len(violations) == 1 + assert "[PAYLOAD_STABLE_UNTIL_READY] output_ cycle 3" in violations[0] diff --git a/docs/src/cocotb.md b/docs/src/cocotb.md index dc86084..43bc880 100644 --- a/docs/src/cocotb.md +++ b/docs/src/cocotb.md @@ -62,9 +62,27 @@ async def test_counting(dut): - **Import order** — `cocotb.handle` must be imported before `cocotb._gpi_triggers` in patched simulator contexts. `run_cocotb` handles this automatically. - **Multi-domain edge semantics** — posedge and negedge domains can share clocks and progress with correct edge-firing behavior. +## Stream contract monitoring + +Attach the additive checker as an async context manager around existing bench stimulus. It samples only clock edges outside reset and reports the interface prefix, cycle, and rule name on failure. + +```python +from dau_sim.integrations.protocol import StreamContractMonitor + + +async with StreamContractMonitor(dut, dut.clk, "output_", reset=dut.rst, expected_batches=1): + await drive_and_drain_one_batch(dut) +``` + +The default payload is `data` plus `last`. A stalled `valid` and its payload must remain stable until `ready`; transfers are counted only when both are high. `expected_batches` makes duplicate or missing `last` assertions decidable and rejects transfers after the final `last`. + +`StatusContractMonitor` applies the same hold rule to `status_valid`, `status_error`, and `status_error_code`. Set `mode="terminal"` to require one status per expected batch or `mode="mid_lane"` to reject success statuses. Both monitors are simulator-neutral cocotb code, so the same bench can use them with dau-sim or Verilator. Neither monitor changes launcher or backend defaults. + ## API | Function / Class | Description | | -------------------------------------- | ----------------------------------------------------------- | | `run_cocotb(design, test_module, ...)` | Run cocotb testbench against Amaranth design or IR `Module` | | `SimulationEngine(module)` | Low-level engine with NBA-correct event scheduling | +| `StreamContractMonitor(...)` | Check a prefixed valid/ready/data/last stream | +| `StatusContractMonitor(...)` | Check terminal or mid-lane status handshakes |