• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

Nic30 / hwtLib / 33a5e7e6-56d8-4169-84e3-820aeb2d5ab4

30 May 2026 09:37PM UTC coverage: 92.767% (+0.04%) from 92.73%
33a5e7e6-56d8-4169-84e3-820aeb2d5ab4

push

circleci

Nic30
fix(package): update MANIFEST.in

5552 of 6796 branches covered (81.7%)

41386 of 44613 relevant lines covered (92.77%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.22
/hwtLib/amba/axi_comp/cache/cacheWriteAllocWawOnlyWritePropagating_test.py
1
#!/usr/bin/env python3
2
# -*- coding: utf-8 -*-
3

4
from hwt.hdl.types.bits import HBits
1✔
5
from hwt.hdl.types.bitsConst import HBitsConst
1✔
6
from hwt.serializer.combLoopAnalyzer import CombLoopAnalyzer
1✔
7
from hwt.simulator.simTestCase import SimTestCase
1✔
8
from hwtLib.amba.axiLite_comp.sim.utils import axi_randomize_per_channel
1✔
9
from hwtLib.amba.axi_comp.cache.cacheWriteAllocWawOnlyWritePropagating import AxiCacheWriteAllocWawOnlyWritePropagating
1✔
10
from hwtLib.amba.constants import RESP_OKAY
1✔
11
from hwtLib.mem.sim.segmentedArrayProxy import SegmentedArrayProxy
1✔
12
from hwtSimApi.constants import CLK_PERIOD
1✔
13
from pyMathBitPrecise.bit_utils import set_bit_range, mask, int_list_to_int
1✔
14

15

16
class AxiCacheWriteAllocWawOnlyWritePropagatingTC(SimTestCase):
1✔
17
    # number of words in transaction - 1
18
    LEN = 0
1✔
19

20
    @classmethod
1✔
21
    def setUpClass(cls):
1✔
22
        cls.dut = dut = AxiCacheWriteAllocWawOnlyWritePropagating()
1✔
23
        dut.IS_PREINITIALIZED = True
1✔
24
        dut.DATA_WIDTH = 32
1✔
25
        dut.CACHE_LINE_SIZE = 4 * (cls.LEN + 1)
1✔
26
        dut.CACHE_LINE_CNT = 16
1✔
27
        dut.MAX_BLOCK_DATA_WIDTH = 8
1✔
28
        dut.WAY_CNT = 2
1✔
29
        cls.ADDR_STEP = dut.CACHE_LINE_SIZE
1✔
30
        cls.WAY_CACHELINES = dut.CACHE_LINE_CNT // dut.WAY_CNT
1✔
31
        cls.compileSim(dut)
1✔
32

33
    def setUp(self):
1✔
34
        SimTestCase.setUp(self)
1✔
35
        m = self.rtl_simulator.model
1✔
36
        dut = self.dut
1✔
37
        self.TAGS = SegmentedArrayProxy([
1✔
38
                getattr(m.tag_array_inst.tag_mem_inst, f"children_{i:d}_inst").io.ram_memory
39
                for i in range(dut.tag_array.tag_record_t.bit_length() * dut.WAY_CNT // 8)
40
            ],
41
        )
42
        self.DATA = SegmentedArrayProxy([
1✔
43
                getattr(m.data_array_inst.data_array_inst, f"children_{i:d}_inst").io.ram_memory
44
                for i in range(dut.DATA_WIDTH // 8)
45
            ],
46
            words_per_item=self.LEN + 1
47
        )
48

49
    def set_data(self, index: int, way: int, data: HBitsConst):
1✔
50
        dut = self.dut
1✔
51
        WAY_CACHELINES = dut.CACHE_LINE_CNT // dut.WAY_CNT
1✔
52
        self.DATA[way * WAY_CACHELINES + index] = data
1✔
53

54
    def build_cacheline(self, cacheline_words):
1✔
55
        return int_list_to_int(cacheline_words, self.dut.DATA_WIDTH)
1✔
56

57
    def get_data(self, index, way):
1✔
58
        return self.DATA[way * self.WAY_CACHELINES + index]
1✔
59

60
    def set_tag(self, addr, way):
1✔
61
        dut = self.dut
1✔
62
        tag, index, offset = dut.parse_addr_int(addr)
1✔
63
        assert offset == 0, addr
1✔
64
        tag_t = dut.tag_array.tag_record_t
1✔
65
        tag_t_w = tag_t.bit_length()
1✔
66
        v = tag_t.from_py({"tag": tag, "valid": 1})._reinterpret_cast(HBits(tag_t_w))
1✔
67

68
        cur_v = self.TAGS[index]
1✔
69
        assert cur_v._is_full_valid(), (cur_v, index)
1✔
70
        val = set_bit_range(cur_v.val, way * tag_t_w, tag_t_w, v.val)
1✔
71
        self.TAGS[index] = val
1✔
72

73
        return index
1✔
74

75
    def cacheline_insert(self, addr: int, way: int, data: HBitsConst):
1✔
76
        index = self.set_tag(addr, way)
1✔
77
        self.set_data(index, way, data)
1✔
78

79
    def get_cachelines(self):
1✔
80
        dut = self.dut
1✔
81
        res = {}
1✔
82
        tags_t = dut.tag_array.tag_record_t[dut.WAY_CNT]
1✔
83
        tags_raw_t = HBits(tags_t.bit_length())
1✔
84
        for index in range(2 ** dut.INDEX_W):
1✔
85
            tags = self.TAGS[index]
1✔
86
            tags = tags_raw_t.from_py(tags.val, tags.vld_mask)._reinterpret_cast(tags_t)
1✔
87
            for way, t in enumerate(tags):
1✔
88
                if t.valid:
1✔
89
                    data = self.get_data(index, way)
1✔
90
                    addr = dut.deparse_addr(t.tag, index, 0)
1✔
91
                    if data._is_full_valid():
1!
92
                        data = int(data)
1✔
93
                    res[int(addr)] = data
1✔
94
        return res
1✔
95

96
    def randomize_all(self):
1✔
97
        dut = self.dut
1✔
98
        axi_randomize_per_channel(self, dut.s)
1✔
99
        axi_randomize_per_channel(self, dut.m)
1✔
100

101
    def test_utils(self):
1✔
102
        self.TAGS.clean()
1✔
103
        self.DATA.clean()
1✔
104
        self.assertDictEqual(self.get_cachelines(), {})
1✔
105

106
        expected = {}
1✔
107
        MAGIC = 99
1✔
108
        for w in range(self.dut.WAY_CNT):
1✔
109
            a = (w * self.WAY_CACHELINES + 3) * self.ADDR_STEP
1✔
110
            self.cacheline_insert(a, w, MAGIC + w)
1✔
111
            expected[a] = MAGIC + w
1✔
112
            self.assertDictEqual(self.get_cachelines(), expected)
1✔
113

114
    def test_no_comb_loops(self):
1✔
115
        CombLoopAnalyzer.check_comb_loops_in_SimTestCase(self)
1✔
116

117
    def test_nop(self):
1✔
118
        dut = self.dut
1✔
119
        self.randomize_all()
1✔
120
        self.runSim(50 * CLK_PERIOD)
1✔
121
        for x in [dut.m.ar, dut.m.aw, dut.m.w, dut.s.r, dut.s.b]:
1✔
122
            self.assertEmpty(x._ag.data)
1✔
123

124
    def test_read_through(self, N=10, randomized=False):
1✔
125
        # Every transaction should be checked if present in cache, should not be found and should
126
        # be forwarded on axi "m" port
127
        dut = self.dut
1✔
128
        self.TAGS.clean()
1✔
129
        ref = [
1✔
130
            dut.s.ar._ag.create_addr_req(addr=i * self.ADDR_STEP, _len=self.LEN, _id=i)
131
            for i in range(N)
132
        ]
133
        dut.s.ar._ag.data.extend(ref)
1✔
134
        t = (N * (self.LEN + 1) + 5) * CLK_PERIOD
1✔
135
        if randomized:
1!
136
            self.randomize_all()
×
137
            t *= 3
×
138

139
        self.runSim(t)
1✔
140
        self.assertValSequenceEqual(dut.m.ar._ag.data, ref)
1✔
141
        for x in [dut.m.aw, dut.m.w, dut.s.r, dut.s.b]:
1✔
142
            self.assertEmpty(x._ag.data)
1✔
143

144
    def test_read_from_everywhere(self, N_PER_WAY=10, MAGIC=99, randomized=False):
1✔
145
        # Every transaction should be read from the cache
146
        dut = self.dut
1✔
147
        self.TAGS.clean()
1✔
148
        self.DATA.clean()
1✔
149
        ID_MAX = 2 ** dut.ID_WIDTH
1✔
150
        WAY_CACHELINES = self.WAY_CACHELINES
1✔
151
        expected_r = []
1✔
152
        for w in range(dut.WAY_CNT):
1✔
153
            for i in range(N_PER_WAY):
1✔
154
                i = i % WAY_CACHELINES
1✔
155
                addr = (i + w * WAY_CACHELINES) * self.ADDR_STEP
1✔
156
                _id = i % ID_MAX
1✔
157
                req = dut.s.ar._ag.create_addr_req(addr=addr, _len=self.LEN, _id=_id)
1✔
158
                dut.s.ar._ag.data.append(req)
1✔
159
                d_list = []
1✔
160
                for w_i in range(self.LEN + 1):
1✔
161
                    d = w * MAGIC + i + w_i
1✔
162
                    expected_r.append((_id, d, RESP_OKAY, int(w_i == self.LEN)))
1✔
163
                    d_list.append(d)
1✔
164

165
                self.cacheline_insert(addr, w, self.build_cacheline(d_list))
1✔
166

167
        t = (10 + dut.WAY_CNT * N_PER_WAY * (self.LEN + 1)) * CLK_PERIOD
1✔
168
        if randomized:
1✔
169
            self.randomize_all()
1✔
170
            t *= 3
1✔
171
        self.runSim(t)
1✔
172
        for x in [dut.s.ar, dut.m.ar, dut.m.aw, dut.m.w, dut.s.b]:
1✔
173
            self.assertEmpty(x._ag.data, x)
1✔
174

175
        self.assertValSequenceEqual(dut.s.r._ag.data, expected_r)
1✔
176

177
    def test_read_from_everywhere_r(self, N_PER_WAY=50, MAGIC=99):
1✔
178
        self.test_read_from_everywhere(
1✔
179
            N_PER_WAY=N_PER_WAY,
180
            MAGIC=MAGIC,
181
            randomized=True)
182

183
    def _test_write_to(self, N_PER_WAY=8, MAGIC=99, preallocate=False, randomized=False):
1✔
184
        # Every cacheline should be stored in cache as there is plenty of space
185
        dut = self.dut
1✔
186
        self.TAGS.clean()
1✔
187
        self.DATA.clean()
1✔
188

189
        WAY_CACHELINES = self.WAY_CACHELINES
1✔
190
        ID_MAX = 2 ** dut.ID_WIDTH
1✔
191
        aw = dut.s.aw._ag
1✔
192
        expected = {}
1✔
193
        b_expected = []
1✔
194
        M = mask(dut.DATA_WIDTH // 8)
1✔
195
        for w in range(dut.WAY_CNT):
1✔
196
            for i in range(N_PER_WAY):
1✔
197
                i = i % WAY_CACHELINES
1✔
198
                addr = (i + w * WAY_CACHELINES) * self.ADDR_STEP
1✔
199

200
                if preallocate:
1✔
201
                    self.cacheline_insert(addr, w, 0)
1✔
202

203
                _id = i % ID_MAX
1✔
204
                req = aw.create_addr_req(addr=addr, _len=self.LEN, _id=_id)
1✔
205
                aw.data.append(req)
1✔
206
                d_list = []
1✔
207
                for w_i in range(self.LEN + 1):
1✔
208
                    d = w * MAGIC + i + w_i
1✔
209
                    dut.s.w._ag.data.append((d, M, int(w_i == self.LEN)))
1✔
210
                    d_list.append(d)
1✔
211
                expected[addr] = self.build_cacheline(d_list)
1✔
212

213
                b_expected.append((_id, RESP_OKAY))
1✔
214
        if preallocate:
1✔
215
            self.assertDictEqual(self.get_cachelines(), {k: 0 for k in expected.keys()})
1✔
216
        t = (dut.WAY_CNT * N_PER_WAY * (self.LEN + 1) + 10) * CLK_PERIOD
1✔
217
        if randomized:
1✔
218
            t *= 3
1✔
219
            self.randomize_all()
1✔
220

221
        self.runSim(t)
1✔
222
        for x in [dut.s.aw, dut.s.r, dut.m.ar, dut.m.aw, dut.m.w]:
1✔
223
            self.assertEmpty(x._ag.data, x)
1✔
224

225
        self.assertDictEqual(self.get_cachelines(), expected)
1✔
226
        self.assertValSequenceEqual(dut.s.b._ag.data, b_expected)
1✔
227

228
    def test_write_to_empty(self, N_PER_WAY=8, MAGIC=99, preallocate=False, randomized=False):
1✔
229
        self._test_write_to(N_PER_WAY, MAGIC, preallocate, randomized)
1✔
230

231
    def test_write_to_preallocated(self, N_PER_WAY=8, MAGIC=99):
1✔
232
        self._test_write_to(N_PER_WAY=N_PER_WAY, MAGIC=MAGIC, preallocate=True)
1✔
233

234
    def test_write_to_empty_r(self, N_PER_WAY=50, MAGIC=99):
1✔
235
        self._test_write_to(N_PER_WAY=N_PER_WAY, MAGIC=MAGIC, preallocate=False, randomized=True)
1✔
236

237
    def test_write_to_preallocated_r(self, N_PER_WAY=50, MAGIC=99):
1✔
238
        self._test_write_to(N_PER_WAY=N_PER_WAY, MAGIC=MAGIC, preallocate=True, randomized=True)
1✔
239

240
    def test_read_write_to_different_sets(self, N_PER_WAY=5, MAGIC=99, randomized=False):
1✔
241
        dut = self.dut
1✔
242
        self.TAGS.clean()
1✔
243
        self.DATA.clean()
1✔
244

245
        WAY_CACHELINES = self.WAY_CACHELINES
1✔
246
        ID_MAX = 2 ** dut.ID_WIDTH
1✔
247
        aw = dut.s.aw._ag
1✔
248
        expected = {}
1✔
249
        b_expected = []
1✔
250
        M = mask(dut.DATA_WIDTH // 8)
1✔
251
        expected_r = []
1✔
252
        for w in range(dut.WAY_CNT):
1✔
253
            for i in range(N_PER_WAY):
1✔
254
                i = i % WAY_CACHELINES
1✔
255
                addr = (i + w * WAY_CACHELINES) * self.ADDR_STEP
1✔
256

257
                _id = i % ID_MAX
1✔
258
                req = aw.create_addr_req(addr=addr, _len=0, _id=_id)
1✔
259

260
                dut.s.ar._ag.data.append(req)
1✔
261

262
                aw.data.append(req)
1✔
263
                r_d_list = []
1✔
264
                w_d_list = []
1✔
265
                for w_i in range(self.LEN + 1):
1✔
266
                    last = int(w_i == self.LEN)
1✔
267

268
                    write_d = (dut.WAY_CNT + w) * MAGIC + i + w_i
1✔
269
                    dut.s.w._ag.data.append((write_d, M, last))
1✔
270
                    w_d_list.append(write_d)
1✔
271

272
                    read_d = w * MAGIC + i + w_i
1✔
273
                    expected_r.append((_id, read_d, RESP_OKAY, last))
1✔
274
                    r_d_list.append(read_d)
1✔
275

276
                expected[addr] = self.build_cacheline(w_d_list)
1✔
277
                self.cacheline_insert(addr, w, self.build_cacheline(r_d_list))
1✔
278

279
                b_expected.append((_id, RESP_OKAY))
1✔
280

281
        t = (dut.WAY_CNT * N_PER_WAY * (self.LEN + 1) + 10) * CLK_PERIOD
1✔
282
        if randomized:
1!
283
            t *= 3
×
284
            self.randomize_all()
×
285

286
        self.runSim(t)
1✔
287
        for x in [dut.s.aw, dut.m.ar, dut.m.aw, dut.m.w]:
1✔
288
            self.assertEmpty(x._ag.data, x)
1✔
289

290
        self.assertDictEqual(self.get_cachelines(), expected)
1✔
291
        self.assertValSequenceEqual(dut.s.b._ag.data, b_expected)
1✔
292
        self.assertValSequenceEqual(dut.s.r._ag.data, expected_r)
1✔
293

294
    # write victim flush
295

296

297
class AxiCacheWriteAllocWawOnlyWritePropagating_len1TC(AxiCacheWriteAllocWawOnlyWritePropagatingTC):
1✔
298
    LEN = 1
1✔
299

300

301
AxiCacheWriteAllocWawOnlyWritePropagatingTCs = [
1✔
302
    AxiCacheWriteAllocWawOnlyWritePropagatingTC,
303
    AxiCacheWriteAllocWawOnlyWritePropagating_len1TC,
304
]
305

306
if __name__ == "__main__":
307
    import unittest
308
    testLoader = unittest.TestLoader()
309
    # suite = unittest.TestSuite([AxiCacheWriteAllocWawOnlyWritePropagatingTC("test_write_to_empty")])
310
    loadedTcs = [testLoader.loadTestsFromTestCase(tc) for tc in AxiCacheWriteAllocWawOnlyWritePropagatingTCs]
311
    suite = unittest.TestSuite(loadedTcs)
312
    runner = unittest.TextTestRunner(verbosity=3)
313
    runner.run(suite)
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc