fix(replay): drive payloads and pulse event inputs

ut_mem_block/lsq/replay_queue/agent/LoadQueueReplayAgent.py

@@ -118,15 +118,21 @@ async def Update_blocking(self, stDataReadySqPtr: ReadySqPtr, stAddrReadySqPtr:
         self.bundle.io._rawFull.value = rawFull
         self.bundle.io._ldWbPtr._flag.value = ldWbPtr.flag
         self.bundle.io._ldWbPtr._value.value = ldWbPtr.value
-        await self.bundle.step(2)
+        await self.bundle.step(1)
+        for i in range(2):
+            getattr(self.bundle.io._storeAddrIn, f'_{i}')._valid.value = False
+            getattr(self.bundle.io._storeDataIn, f'_{i}')._valid.value = False
+        self.bundle.io._tlb_hint_resp._valid.value = False
+        self.bundle.io._tl_d_channel._valid.value = False
+        await self.bundle.step(1)
         return self.bundle.LoadQueueReplay
 
     @driver_method()
     async def replay(self, l2_hint: L2Hint):
         self.bundle.io._l2_hint._valid.value = l2_hint.valid
-        self.bundle.io._l2_hint._bits._sourceId = l2_hint.sourceId
-        self.bundle.io._l2_hint._bits._isKeyword = l2_hint.isKeyword
-        await self.bundle.step(4)
+        self.bundle.io._l2_hint._bits._sourceId.value = l2_hint.sourceId
+        self.bundle.io._l2_hint._bits._isKeyword.value = l2_hint.isKeyword
+        await self.bundle.step(1)
+        self.bundle.io._l2_hint._valid.value = False
+        await self.bundle.step(3)
         return self.bundle.LoadQueueReplay._scheduled
-
-

ut_mem_block/lsq/replay_queue/test/test_agent_event_inputs.py

@@ -0,0 +1,150 @@
+# coding=utf8
+# ***************************************************************************************
+# This project is licensed under Mulan PSL v2.
+# You can use this software according to the terms and conditions of the Mulan PSL v2.
+# You may obtain a copy of Mulan PSL v2 at:
+#          http://license.coscl.org.cn/MulanPSL2
+#
+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+#
+# See the Mulan PSL v2 for more details.
+# **************************************************************************************/
+
+
+import asyncio
+from types import SimpleNamespace
+
+from ..agent.LoadQueueReplayAgent import LoadQueueReplayAgent
+from ..util.dataclass import (
+    IOldWbPtr,
+    ReadySqPtr,
+    StoreAddrIn,
+    StoreDataIn,
+    TLChannel,
+    TlbHint,
+)
+
+
+class RecordingPin:
+    def __init__(self):
+        self.writes = []
+
+    @property
+    def value(self):
+        return self.writes[-1] if self.writes else None
+
+    @value.setter
+    def value(self, new_value):
+        self.writes.append(new_value)
+
+
+def pin_vector(size):
+    return SimpleNamespace(**{f"_{index}": RecordingPin() for index in range(size)})
+
+
+def pointer():
+    return SimpleNamespace(_flag=RecordingPin(), _value=RecordingPin())
+
+
+def store_addr_port():
+    return SimpleNamespace(
+        _valid=RecordingPin(),
+        _bits=SimpleNamespace(_uop_sqIdx=pointer(), _miss=RecordingPin()),
+    )
+
+
+def store_data_port():
+    return SimpleNamespace(_valid=RecordingPin(), _bits_uop_sqIdx=pointer())
+
+
+class RecordingBundle:
+    def __init__(self):
+        self.store_addr_ports = [store_addr_port() for _ in range(2)]
+        self.store_data_ports = [store_data_port() for _ in range(2)]
+        self.io = SimpleNamespace(
+            _stDataReadySqPtr=pointer(),
+            _sqEmpty=RecordingPin(),
+            _stAddrReadyVec=pin_vector(56),
+            _stDataReadyVec=pin_vector(56),
+            _storeAddrIn=SimpleNamespace(
+                **{f"_{index}": port for index, port in enumerate(self.store_addr_ports)}
+            ),
+            _storeDataIn=SimpleNamespace(
+                **{f"_{index}": port for index, port in enumerate(self.store_data_ports)}
+            ),
+            _stAddrReadySqPtr=pointer(),
+            _tlb_hint_resp=SimpleNamespace(
+                _valid=RecordingPin(),
+                _bits=SimpleNamespace(_id=RecordingPin(), _replay_all=RecordingPin()),
+            ),
+            _tl_d_channel=SimpleNamespace(_valid=RecordingPin(), _mshrid=RecordingPin()),
+            _rarFull=RecordingPin(),
+            _rawFull=RecordingPin(),
+            _ldWbPtr=pointer(),
+        )
+        self.LoadQueueReplay = object()
+        self.step_calls = []
+
+    async def step(self, cycles):
+        self.step_calls.append(cycles)
+
+
+def test_update_blocking_pulses_event_valids_and_preserves_level_inputs():
+    bundle = RecordingBundle()
+    agent = object.__new__(LoadQueueReplayAgent)
+    agent.bundle = bundle
+
+    update_blocking = LoadQueueReplayAgent.Update_blocking.__original_func__
+    result = asyncio.run(
+        update_blocking(
+            agent,
+            stDataReadySqPtr=ReadySqPtr(flag=True, value=9),
+            stAddrReadySqPtr=ReadySqPtr(flag=False, value=7),
+            sqEmpty=True,
+            storeAddrIn=[
+                StoreAddrIn(valid=True, sqIdx_flag=True, sqIdx_value=3, miss=False),
+                StoreAddrIn(valid=True, sqIdx_flag=False, sqIdx_value=4, miss=True),
+            ],
+            storeDataIn=[
+                StoreDataIn(valid=True, sqIdx_flag=True, sqIdx_value=5),
+                StoreDataIn(valid=True, sqIdx_flag=False, sqIdx_value=6),
+            ],
+            stAddrReadyVec=[True] * 56,
+            stDataReadyVec=[False] * 56,
+            tlb_hint=TlbHint(valid=True, id=11, replay_all=False),
+            tl_channel=TLChannel(valid=True, mshrid=13),
+            rarFull=True,
+            ldWbPtr=IOldWbPtr(flag=True, value=15),
+            rawFull=True,
+        )
+    )
+
+    event_valids = [
+        *(port._valid for port in bundle.store_addr_ports),
+        *(port._valid for port in bundle.store_data_ports),
+        bundle.io._tlb_hint_resp._valid,
+        bundle.io._tl_d_channel._valid,
+    ]
+    assert all(pin.writes == [True, False] for pin in event_valids)
+    assert all(pin.value is False for pin in event_valids)
+    assert bundle.io._sqEmpty.value is True
+    assert bundle.io._rarFull.value is True
+    assert bundle.io._rawFull.value is True
+    assert bundle.io._stDataReadySqPtr._flag.value is True
+    assert bundle.io._stDataReadySqPtr._value.value == 9
+    assert bundle.io._stAddrReadySqPtr._flag.value is False
+    assert bundle.io._stAddrReadySqPtr._value.value == 7
+    assert bundle.io._ldWbPtr._flag.value is True
+    assert bundle.io._ldWbPtr._value.value == 15
+    assert all(
+        getattr(bundle.io._stAddrReadyVec, f"_{index}").value is True
+        for index in range(56)
+    )
+    assert all(
+        getattr(bundle.io._stDataReadyVec, f"_{index}").value is False
+        for index in range(56)
+    )
+    assert bundle.step_calls == [1, 1]
+    assert result is bundle.LoadQueueReplay

ut_mem_block/lsq/replay_queue/test/test_agent_l2_hint.py

@@ -0,0 +1,74 @@
+# coding=utf8
+# ***************************************************************************************
+# This project is licensed under Mulan PSL v2.
+# You can use this software according to the terms and conditions of the Mulan PSL v2.
+# You may obtain a copy of Mulan PSL v2 at:
+#          http://license.coscl.org.cn/MulanPSL2
+#
+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+#
+# See the Mulan PSL v2 for more details.
+# **************************************************************************************/
+
+
+import asyncio
+from types import SimpleNamespace
+
+from ..agent.LoadQueueReplayAgent import LoadQueueReplayAgent
+from ..util.dataclass import L2Hint
+
+
+class RecordingPin:
+    def __init__(self):
+        self.writes = []
+
+    @property
+    def value(self):
+        return self.writes[-1] if self.writes else None
+
+    @value.setter
+    def value(self, new_value):
+        self.writes.append(new_value)
+
+
+class RecordingBundle:
+    def __init__(self, scheduled):
+        self.valid_pin = RecordingPin()
+        self.source_id_pin = RecordingPin()
+        self.is_keyword_pin = RecordingPin()
+        hint_bits = SimpleNamespace(
+            _sourceId=self.source_id_pin,
+            _isKeyword=self.is_keyword_pin,
+        )
+        self.io = SimpleNamespace(
+            _l2_hint=SimpleNamespace(_valid=self.valid_pin, _bits=hint_bits)
+        )
+        self.LoadQueueReplay = SimpleNamespace(_scheduled=scheduled)
+        self.step_calls = []
+
+    async def step(self, cycles):
+        self.step_calls.append(cycles)
+
+
+def test_replay_drives_l2_hint_payload_and_pulses_valid():
+    scheduled = object()
+    bundle = RecordingBundle(scheduled)
+    source_id_pin = bundle.source_id_pin
+    is_keyword_pin = bundle.is_keyword_pin
+    agent = object.__new__(LoadQueueReplayAgent)
+    agent.bundle = bundle
+    l2_hint = L2Hint(valid=True, sourceId=7, isKeyword=False)
+
+    replay = LoadQueueReplayAgent.replay.__original_func__
+    result = asyncio.run(replay(agent, l2_hint))
+
+    assert bundle.valid_pin.writes == [True, False]
+    assert bundle.valid_pin.value is False
+    assert bundle.io._l2_hint._bits._sourceId is source_id_pin
+    assert bundle.io._l2_hint._bits._isKeyword is is_keyword_pin
+    assert source_id_pin.writes == [7]
+    assert is_keyword_pin.writes == [False]
+    assert bundle.step_calls == [1, 3]
+    assert result is scheduled