diff --git a/ut_frontend/bpu/ittage/env/ittage_wrapper.py b/ut_frontend/bpu/ittage/env/ittage_wrapper.py index e6e3d8d5..b677d115 100644 --- a/ut_frontend/bpu/ittage/env/ittage_wrapper.py +++ b/ut_frontend/bpu/ittage/env/ittage_wrapper.py @@ -1,19 +1,28 @@ import os.path import random from datetime import datetime +from typing import TYPE_CHECKING import toffee from toffee import ClockCycles -from dut.ITTage import DUTITTage from ..util.common import get_folded_hist from .bundle import * +if TYPE_CHECKING: + from dut.ITTage import DUTITTage + __all__ = ['ITTageWrapper'] +# The current generated RTL clears five 128-entry tables in parallel and was +# observed ready 130 cycles after reset deassertion. Keep ample headroom for +# wrapper/RTL evolution without restoring an unbounded wait. +DEFAULT_READY_TIMEOUT_CYCLES = 1024 + + class ITTageWrapper: - def __init__(self, dut: DUTITTage): + def __init__(self, dut: "DUTITTage", ready_timeout_cycles: int = DEFAULT_READY_TIMEOUT_CYCLES): # Create DUT self.dut = dut self.dut.InitClock("clock") @@ -26,7 +35,7 @@ def __init__(self, dut: DUTITTage): self.pipeline_ctrl = PipelineCtrl.from_prefix("io_").set_name("ittage_pipeline_ctrl").bind(self.dut) # Reset - self.reset() + self.reset(ready_timeout_cycles) def finalize(self): self.dut.Finish() @@ -122,13 +131,25 @@ def update(self, update_req): def update_async(self, update_req): return toffee.create_task(self.__update_async__(update_req)) - def reset(self): + def reset(self, ready_timeout_cycles: int = DEFAULT_READY_TIMEOUT_CYCLES): + if ready_timeout_cycles <= 0: + raise ValueError("ready_timeout_cycles must be greater than zero") + self.dut.reset.value = 1 self.xclock.Step(1) self.dut.reset.value = 0 + + waited_cycles = 0 while self.dut.io_s1_ready.value == 0: + if waited_cycles >= ready_timeout_cycles: + raise TimeoutError( + "ITTage reset timed out after " + f"{waited_cycles} cycles waiting for io_s1_ready=1 " + f"(last value: {self.dut.io_s1_ready.value})" + ) self.xclock.Step(1) + waited_cycles += 1 self.xclock.Step(10) def reset_async(self): - return toffee.create_task(self.__rst_async__()) \ No newline at end of file + return toffee.create_task(self.__rst_async__()) diff --git a/ut_frontend/bpu/ittage/test/test_reset_timeout.py b/ut_frontend/bpu/ittage/test/test_reset_timeout.py new file mode 100644 index 00000000..b069458c --- /dev/null +++ b/ut_frontend/bpu/ittage/test/test_reset_timeout.py @@ -0,0 +1,109 @@ +import pytest + +from ..env import ittage_wrapper as wrapper_module + + +class FakeSignal: + def __init__(self, value): + self.value = value + + +class FakeClock: + def __init__(self, dut, ready_after_wait_cycles): + self.dut = dut + self.ready_after_wait_cycles = ready_after_wait_cycles + self.step_calls = [] + self.waited_cycles = 0 + + def Step(self, cycles): + self.step_calls.append(cycles) + if self.dut.reset.value == 0 and self.dut.io_s1_ready.value == 0: + self.waited_cycles += cycles + if ( + self.ready_after_wait_cycles is not None + and self.waited_cycles >= self.ready_after_wait_cycles + ): + self.dut.io_s1_ready.value = 1 + + +class FakeDut: + def __init__(self, ready_after_wait_cycles): + initially_ready = ready_after_wait_cycles == 0 + self.reset = FakeSignal(0) + self.io_s1_ready = FakeSignal(int(initially_ready)) + self.xclock = FakeClock(self, ready_after_wait_cycles) + self.clock_names = [] + + def InitClock(self, name): + self.clock_names.append(name) + + +class FakeBundle: + @classmethod + def from_prefix(cls, _prefix): + return cls() + + def set_name(self, _name): + return self + + def bind(self, _dut): + return self + + +def make_wrapper(ready_after_wait_cycles): + dut = FakeDut(ready_after_wait_cycles) + wrapper = wrapper_module.ITTageWrapper.__new__(wrapper_module.ITTageWrapper) + wrapper.dut = dut + wrapper.xclock = dut.xclock + return wrapper, dut + + +def test_reset_returns_immediately_when_ready(): + wrapper, dut = make_wrapper(ready_after_wait_cycles=0) + + wrapper.reset(ready_timeout_cycles=3) + + assert dut.xclock.step_calls == [1, 10] + + +@pytest.mark.parametrize("ready_timeout_cycles", [0, -1]) +def test_reset_rejects_non_positive_timeout_without_clocking(ready_timeout_cycles): + wrapper, dut = make_wrapper(ready_after_wait_cycles=None) + + with pytest.raises(ValueError, match="must be greater than zero"): + wrapper.reset(ready_timeout_cycles=ready_timeout_cycles) + + assert dut.reset.value == 0 + assert dut.xclock.step_calls == [] + + +def test_reset_accepts_ready_on_last_allowed_cycle(): + wrapper, dut = make_wrapper(ready_after_wait_cycles=3) + + wrapper.reset(ready_timeout_cycles=3) + + assert dut.xclock.waited_cycles == 3 + assert dut.xclock.step_calls == [1, 1, 1, 1, 10] + + +def test_reset_times_out_after_bounded_wait(): + wrapper, dut = make_wrapper(ready_after_wait_cycles=None) + + with pytest.raises(TimeoutError, match=r"3 cycles waiting for io_s1_ready=1"): + wrapper.reset(ready_timeout_cycles=3) + + assert dut.reset.value == 0 + assert dut.xclock.waited_cycles == 3 + assert dut.xclock.step_calls == [1, 1, 1, 1] + + +def test_constructor_forwards_configurable_timeout(monkeypatch): + for bundle_name in ("UpdateBundle", "InBundle", "OutBundle", "PipelineCtrl"): + monkeypatch.setattr(wrapper_module, bundle_name, FakeBundle) + dut = FakeDut(ready_after_wait_cycles=None) + + with pytest.raises(TimeoutError, match=r"2 cycles waiting for io_s1_ready=1"): + wrapper_module.ITTageWrapper(dut, ready_timeout_cycles=2) + + assert dut.clock_names == ["clock"] + assert dut.xclock.step_calls == [1, 1, 1]