From 1870fdf2b0dacf20a83b5c693deb2ba4f0abcd2a Mon Sep 17 00:00:00 2001 From: Florent Kermarrec Date: Tue, 14 Jul 2026 09:38:19 +0200 Subject: [PATCH 1/4] analyzer: fix trigger term flushing and register the comparator operands Two reliability fixes in the trigger: - Term flushing used a fixed 2*depth-cycle window that also ran from reset and kept draining after the memory was already empty. Terms pushed within that window - right after reset, or right after disarming to re-arm - were silently consumed, so the capture armed with an empty term memory and triggered immediately on anything. Slow interactive/UART hosts never noticed; fast CSR transports (PCIe, Ethernet) hit this easily, which shows up as captures that 'do not trigger correctly'. Flush now starts on the falling edge of enable and stops as soon as the memory is empty. - The comparator was a fully-combinational data_width-wide compare fed on one side by the unbuffered AsyncFIFO's block-RAM read port and on the other by the raw probe mux, feeding back into the consume handshake within a single scope cycle - a timing hazard that grows with probe width. The term memory is now a buffered FIFO and the probe data/valid are registered before the compare, so the wide comparator sits between registers. The registered input delays data and hit together by one scope cycle: captures shift by one cycle in absolute time, the trigger position within the capture is unchanged. The first analyzer test relied on the reset-window flush eating its term (it never enabled the trigger): it now arms the trigger properly on a reachable value and additionally checks the trigger sample's position; the group-mux test now observes the match exactly at index 0 with offset=0. --- litescope/core.py | 42 +++++++++++++++++++++++++++++++++--------- test/test_analyzer.py | 18 ++++++++++++++---- 2 files changed, 47 insertions(+), 13 deletions(-) diff --git a/litescope/core.py b/litescope/core.py index db7e010..aa47a43 100644 --- a/litescope/core.py +++ b/litescope/core.py @@ -67,7 +67,11 @@ def __init__(self, data_width, depth=16): self.specials += MultiReg(done, self.done.status) # Memory and configuration. - mem = stream.AsyncFIFO([("mask", data_width), ("value", data_width)], depth) + # Use a buffered AsyncFIFO so Mask/Value come from registers: the unbuffered FIFO's + # output is the block RAM read port, whose clock-to-output delay would otherwise feed + # the data_width-wide comparator and the consume handshake within a single scope cycle + # and break timing as probes get wide. + mem = stream.AsyncFIFO([("mask", data_width), ("value", data_width)], depth, buffered=True) mem = ClockDomainsRenamer({"write": "sys", "read": "scope"})(mem) self.submodules += mem self.comb += [ @@ -77,20 +81,40 @@ def __init__(self, data_width, depth=16): self.mem_full.status.eq(~mem.sink.ready) ] + # Register Sink Data/Valid so the data_width-wide comparator sits between registers. The + # probe stream is free-running (the Mux always asserts valid), so sampling it + # unconditionally delays Data and Hit together by one cycle and preserves their + # alignment (the trigger position within the capture is unchanged). + sink_d = stream.Endpoint(core_layout(data_width)) + self.sync.scope += [ + sink_d.valid.eq(sink.valid), + sink_d.data.eq(sink.data), + ] + self.comb += sink.ready.eq(1) + # Hit and memory read/flush. - hit = Signal() - flush = WaitTimer(2*depth) - flush = ClockDomainsRenamer("scope")(flush) - self.submodules += flush + # Flush pending terms on the falling edge of enable and stop as soon as the memory is + # empty: a fixed-duration flush window (previous implementation) also ran from reset and + # kept draining after the memory was already empty, silently consuming terms pushed + # shortly after reset or re-arming - the capture then started with an empty term memory, + # i.e. an immediate spurious trigger (easily hit with fast CSR transports). + hit = Signal() + flushing = Signal() + self.sync.scope += [ + If(~enable & enable_d, + flushing.eq(1) + ).Elif(~mem.source.valid, + flushing.eq(0) + ) + ] self.comb += [ - flush.wait.eq(~(~enable & enable_d)), # flush when disabling - hit.eq((sink.data & mem.source.mask) == (mem.source.value & mem.source.mask)), - mem.source.ready.eq((enable & hit) | ~flush.done), + hit.eq((sink_d.data & mem.source.mask) == (mem.source.value & mem.source.mask)), + mem.source.ready.eq((enable & sink_d.valid & hit) | flushing), ] # Output. self.comb += [ - sink.connect(source), + sink_d.connect(source, omit={"ready"}), # Done when all triggers have been consumed. done.eq(~mem.source.valid), source.hit.eq(done) diff --git a/test/test_analyzer.py b/test/test_analyzer.py index 2c1067b..8b36669 100644 --- a/test/test_analyzer.py +++ b/test/test_analyzer.py @@ -33,8 +33,10 @@ class TestAnalyzer(unittest.TestCase): def test_analyzer(self): def generator(dut): dut.data = [] - # Configure Trigger - yield from dut.analyzer.trigger.mem_value.write(0x0010) + # Configure Trigger (on a counter value comfortably after arming completes; the + # term memory no longer accepts being armed with pending terms silently flushed, + # so the trigger must actually be enabled and reachable). + yield from dut.analyzer.trigger.mem_value.write(0x0400) yield from dut.analyzer.trigger.mem_mask.write(0xffff) yield from dut.analyzer.trigger.mem_write.write(1) @@ -45,6 +47,7 @@ def generator(dut): yield from dut.analyzer.storage.length.write(256) yield from dut.analyzer.storage.offset.write(8) yield from dut.analyzer.storage.enable.write(1) + yield from dut.analyzer.trigger.enable.write(1) yield for i in range(16): yield @@ -64,7 +67,11 @@ def __init__(self): generators = {"sys" : [generator(dut)]} clocks = {"sys": 10, "scope": 10} run_simulation(dut, generators, clocks, vcd_name="sim.vcd") - self.assertEqual(dut.data, [524 + 3*i for i in range(len(dut.data))]) + # Trigger value 0x400 (comfortably after the storage FLUSH window at depth=512) with + # offset=8 and subsampling=3: the capture holds the pre-trigger window followed by the + # match (0x400 at index 6) and the post-trigger samples. + self.assertEqual(dut.data, [1006 + 3*i for i in range(len(dut.data))]) + self.assertEqual(dut.data.index(0x400), 6) def test_analyzer_group_mux(self): def generator(dut): @@ -105,7 +112,10 @@ def __init__(self): generators = {"sys" : [generator(dut)]} clocks = {"sys": 10, "scope": 10} run_simulation(dut, generators, clocks) - self.assertEqual(dut.data, [132 + 3*i for i in range(len(dut.data))]) + # With offset=0 the trigger match is the first captured sample: 0xb0 at index 0 + # (the second group's signal, proving the mux selection), incrementing by 3. + self.assertEqual(dut.data[0], 0xb0) + self.assertEqual(dut.data, [(0xb0 + 3*i) & 0xff for i in range(len(dut.data))]) def test_analyzer_raw_msb_data_without_rle(self): def generator(dut): From 4dfcbebde1f94bf177252938095a4b46a96bd5eb Mon Sep 17 00:00:00 2001 From: Florent Kermarrec Date: Tue, 14 Jul 2026 09:38:42 +0200 Subject: [PATCH 2/4] analyzer: drain stale read-path samples during storage flush Samples from a previous capture or idle drain can linger in the CDC FIFO and read-width converter between the storage FIFO and the mem_data CSR; the next upload then starts with stale words and every following sample is shifted. Drain the read path during the storage FLUSH state (no new data enters it there), so an upload always starts with the new capture's first sample. --- litescope/core.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/litescope/core.py b/litescope/core.py index aa47a43..4038d25 100644 --- a/litescope/core.py +++ b/litescope/core.py @@ -331,6 +331,13 @@ def __init__(self, data_width, depth): mem_flush = ClockDomainsRenamer("scope")(mem_flush) self.submodules += mem_flush + # Read-path flush: samples from a previous capture/drain can linger in the CDC FIFO and + # read converter; drain them during FLUSH (no new data enters the read path in that + # state) so the upload starts with the new capture's first sample. + flush_read = Signal() + flush_read_sys = Signal() + self.specials += MultiReg(flush_read, flush_read_sys) + # FSM. fsm = FSM(reset_state="IDLE") fsm = ClockDomainsRenamer("scope")(fsm) @@ -347,6 +354,7 @@ def __init__(self, data_width, depth): sink.ready.eq(1), mem_flush.wait.eq(1), mem.source.ready.eq(1), + flush_read.eq(1), If(mem_flush.done, NextState("WAIT") ) @@ -381,7 +389,7 @@ def __init__(self, data_width, depth): self.comb += cdc.source.connect(read_source) self.comb += [ - read_source.ready.eq(self.mem_data.rd_stb | ~self.enable.storage), + read_source.ready.eq(self.mem_data.rd_stb | ~self.enable.storage | flush_read_sys), self.mem_data.status.eq(read_source.data) ] From d60534868075809758418dd0ab9fa2d874fdc735 Mon Sep 17 00:00:00 2001 From: Florent Kermarrec Date: Tue, 14 Jul 2026 09:52:23 +0200 Subject: [PATCH 3/4] driver: fix re-running captures, resolve offsets per group Three driver fixes so a capture can be re-run on the same driver instance: - The gateware's trigger memory consumes its terms on every capture, so re-running armed with an empty memory and triggered immediately on anything. add_trigger and the edge helpers now record the configured terms and run() disarms (letting the gateware flush leftovers), reloads all terms and rearms. As a consequence add_trigger no longer writes the hardware directly; the memory-full check happens at load time. - Storage arms on the rising edge of its enable, but run() only ever wrote 1: a second run() on the same instance never re-armed storage and the previous capture's state was re-read as a new one. run() now clears and rises enable, after reloading the trigger terms: an armed trigger with a consumed term memory outputs a constant hit level that would otherwise fire the re-armed storage immediately. - Signal offsets/masks for trigger conditions were flat attributes overwritten by each group, so a signal present in several groups always resolved to the last group's position regardless of the selected group. Conditions now resolve in the currently selected group; the flat attributes remain for compatibility and an ambiguity warning is printed when they differ across groups. --- litescope/software/driver/analyzer.py | 91 +++++++++++++++++++++------ test/test_driver.py | 76 +++++++++++++++++----- 2 files changed, 131 insertions(+), 36 deletions(-) diff --git a/litescope/software/driver/analyzer.py b/litescope/software/driver/analyzer.py index c0d5b98..78a001c 100644 --- a/litescope/software/driver/analyzer.py +++ b/litescope/software/driver/analyzer.py @@ -71,6 +71,10 @@ def __init__(self, regs, name, config_csv=None, debug=False): self.offset = 0 self.length = None + # Configured trigger terms (mask, value); loaded into the gateware on each run() since + # the trigger memory consumes its terms on every capture. + self.trigger_terms = [] + # Disable trigger and storage self.trigger_enable.write(0) self.storage_enable.write(0) @@ -111,24 +115,45 @@ def build(self): if self.name == key[:len(self.name)]: key = key.replace(self.name + "_", "") setattr(self, key, value) - for signals in self.layouts.values(): + # Group-aware signal offsets/masks: triggers must resolve a signal's position in the + # currently selected group (a signal present in several groups can sit at different + # positions in each). + self.signal_offsets = {} + self.signal_masks = {} + for group, signals in self.layouts.items(): value = 1 for name, length in signals: - setattr(self, name + "_o", value) + self.signal_offsets[(group, name)] = value value = value*(2**length) - for signals in self.layouts.values(): value = 0 for name, length in signals: - setattr(self, name + "_m", (2**length-1) << value) + self.signal_masks[(group, name)] = (2**length-1) << value value += length + # Keep the flat _o/_m attributes for compatibility; warn when a signal is + # ambiguous across groups (the flat attribute then reflects the last group only). + ambiguous = set() + for (group, name), offset in self.signal_offsets.items(): + mask = self.signal_masks[(group, name)] + if hasattr(self, name + "_o") and \ + ((getattr(self, name + "_o") != offset) or (getattr(self, name + "_m") != mask)): + ambiguous.add(name) + setattr(self, name + "_o", offset) + setattr(self, name + "_m", mask) + for name in sorted(ambiguous): + print(f"{self.name}: warning: signal '{name}' is present in several groups at " + "different positions; triggers resolve it in the selected group.") + + def _signal_offset(self, name): + return self.signal_offsets.get((self.group, name), getattr(self, name + "_o")) + + def _signal_mask(self, name): + return self.signal_masks.get((self.group, name), getattr(self, name + "_m")) def configure_group(self, value): self.group = value self.mux_value.write(value) def add_trigger(self, value=0, mask=0, cond=None): - if self.trigger_mem_full.read(): - raise ValueError("Trigger memory full, too much conditions") if cond is not None: for k, v in cond.items(): # Check for binary/hexa expressions @@ -145,23 +170,41 @@ def add_trigger(self, value=0, mask=0, cond=None): if c != "x": v |= int(c, 16 if mx is not None else 2 ) m |= 0xf if mx is not None else 0b1 - value |= getattr(self, k + "_o")*v - mask |= getattr(self, k + "_m") & (getattr(self, k + "_o")*m) + value |= self._signal_offset(k)*v + mask |= self._signal_mask(k) & (self._signal_offset(k)*m) # Else convert to int else: - value |= getattr(self, k + "_o")*int(v, 0) - mask |= getattr(self, k + "_m") - self.trigger_mem_mask.write(mask) - self.trigger_mem_value.write(value) - self.trigger_mem_write.write(1) + value |= self._signal_offset(k)*int(v, 0) + mask |= self._signal_mask(k) + self.trigger_terms.append((mask, value)) def add_rising_edge_trigger(self, name): - self.add_trigger(getattr(self, name + "_o")*0, getattr(self, name + "_m")) - self.add_trigger(getattr(self, name + "_o")*1, getattr(self, name + "_m")) + self.add_trigger(self._signal_offset(name)*0, self._signal_mask(name)) + self.add_trigger(self._signal_offset(name)*1, self._signal_mask(name)) def add_falling_edge_trigger(self, name): - self.add_trigger(getattr(self, name + "_o")*1, getattr(self, name + "_m")) - self.add_trigger(getattr(self, name + "_o")*0, getattr(self, name + "_m")) + self.add_trigger(self._signal_offset(name)*1, self._signal_mask(name)) + self.add_trigger(self._signal_offset(name)*0, self._signal_mask(name)) + + def _load_trigger_terms(self, timeout=1.0): + # Disarm the trigger; the gateware flushes any leftover terms (previous capture, + # aborted sequence) on the falling edge of enable. + self.trigger_enable.write(0) + # The flush window is 2*trigger_depth scope cycles; any CSR access takes far longer, + # but poll done (trigger memory empty) so the reload below cannot race the flush. + deadline = time.time() + timeout + for _ in range(2): + while not self.trigger_done.read(): + if time.time() > deadline: + raise TimeoutError("Trigger memory flush timeout") + # (Re-)load the configured terms so a capture can be re-run without reconfiguring: + # the trigger memory consumes its terms on every capture. + for mask, value in self.trigger_terms: + if self.trigger_mem_full.read(): + raise ValueError("Trigger memory full, too much conditions") + self.trigger_mem_mask.write(mask) + self.trigger_mem_value.write(value) + self.trigger_mem_write.write(1) def configure_trigger(self, value=0, mask=0, cond=None): self.add_trigger(value, mask, cond) @@ -193,8 +236,16 @@ def run(self, offset=0, length=None): self.length = length if self.debug: self._log(f"run (offset={offset}, length={length})") + # Disarm the trigger and reload its terms first: a previous capture leaves the trigger + # armed with an empty (fully consumed) term memory, whose hit output is a constant + # level that would otherwise fire the re-armed storage immediately. With the terms + # reloaded, hit stays low until a real match. + self._load_trigger_terms() self.storage_offset.write(offset) self.storage_length.write(length) + # Storage arms on the rising edge of enable: clear it first so run() also re-arms + # after a previous capture on the same driver instance. + self.storage_enable.write(0) self.storage_enable.write(1) self.trigger_enable.write(1) @@ -203,6 +254,7 @@ def clear(self): self.offset = 0 self.length = None self.rle_enabled = False + self.trigger_terms = [] self.trigger_enable.write(0) self.storage_enable.write(0) if hasattr(self, "rle_enable"): @@ -321,11 +373,12 @@ def get_instant_value(self, group, name): self.data = DumpData(self.data_width) self.debug = False self.configure_group(group) + self.trigger_terms = [] self.configure_trigger() self.configure_subsampler(1) self.run(0, 1) self.wait_done() self.upload() - min_idx = log2_int(getattr(self, name + "_o")) - max_idx = min_idx + log2_int((getattr(self, name + "_m") >> min_idx) + 1) + min_idx = log2_int(self._signal_offset(name)) + max_idx = min_idx + log2_int((self._signal_mask(name) >> min_idx) + 1) return self.data[min_idx:max_idx][0] diff --git a/test/test_driver.py b/test/test_driver.py index e435f63..9e7d0cc 100644 --- a/test/test_driver.py +++ b/test/test_driver.py @@ -70,6 +70,7 @@ def make_regs(name="analyzer", mem_level=0, mem_data=None, with_rle=False): "trigger_mem_value": FakeReg(), "trigger_mem_write": FakeReg(), "trigger_enable": FakeReg(), + "trigger_done": FakeReg(1), "subsampler_value": FakeReg(), "storage_offset": FakeReg(), "storage_length": FakeReg(), @@ -203,37 +204,36 @@ def test_conditional_trigger_parsing(self): self.clear_writes(regs) driver.add_trigger(cond={"flag": "1", "state": "0b1x0"}) + self.assertEqual(driver.trigger_terms, [(0xb, 0x9)]) - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0x9]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0xb]) - self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1]) - - self.clear_writes(regs) driver.add_trigger(cond={"wide": "0xax"}) + self.assertEqual(driver.trigger_terms[1], (0xf0, 0xa0)) - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0xa0]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0xf0]) - self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1]) + # Terms are written to the gateware when (re)loaded, not when added. + self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, []) + driver._load_trigger_terms() + self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0x9, 0xa0]) + self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0xb, 0xf0]) + self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1, 1]) def test_edge_trigger_helpers(self): driver, regs = self.make_driver() self.clear_writes(regs) driver.add_rising_edge_trigger("flag") - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0, 1]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [1, 1]) + self.assertEqual(driver.trigger_terms, [(1, 0), (1, 1)]) - self.clear_writes(regs) + driver.trigger_terms = [] driver.add_falling_edge_trigger("flag") - self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [1, 0]) - self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [1, 1]) + self.assertEqual(driver.trigger_terms, [(1, 1), (1, 0)]) - def test_add_trigger_checks_memory_full(self): + def test_load_checks_memory_full(self): driver, regs = self.make_driver() regs.d["analyzer_trigger_mem_full"].value = 1 + driver.add_trigger(value=1, mask=1) with self.assertRaises(ValueError): - driver.add_trigger(value=1, mask=1) + driver._load_trigger_terms() def test_configure_subsampler_and_run(self): driver, regs = self.make_driver(depth=8, subsampler_width=4) @@ -250,14 +250,56 @@ def test_configure_subsampler_and_run(self): self.assertEqual(regs.d["analyzer_subsampler_value"].writes, [3, 15]) self.assertEqual(regs.d["analyzer_storage_offset"].writes, [2]) self.assertEqual(regs.d["analyzer_storage_length"].writes, [5]) - self.assertEqual(regs.d["analyzer_storage_enable"].writes, [1]) - self.assertEqual(regs.d["analyzer_trigger_enable"].writes, [1]) + # run() re-arms storage (rising edge) and disarms/reloads/rearms the trigger. + self.assertEqual(regs.d["analyzer_storage_enable"].writes, [0, 1]) + self.assertEqual(regs.d["analyzer_trigger_enable"].writes, [0, 1]) with self.assertRaises(AssertionError): driver.run(offset=8) with self.assertRaises(AssertionError): driver.run(length=9) + def test_run_reloads_trigger_terms(self): + driver, regs = self.make_driver() + self.clear_writes(regs) + + driver.add_trigger(value=0x3, mask=0x7) + driver.add_trigger(value=0x1, mask=0x1) + + driver.run(offset=0, length=8) + driver.run(offset=0, length=8) + + # Both terms are loaded on each run so captures can be re-run without reconfiguring. + self.assertEqual(regs.d["analyzer_trigger_mem_value"].writes, [0x3, 0x1, 0x3, 0x1]) + self.assertEqual(regs.d["analyzer_trigger_mem_mask"].writes, [0x7, 0x1, 0x7, 0x1]) + self.assertEqual(regs.d["analyzer_trigger_mem_write"].writes, [1, 1, 1, 1]) + self.assertEqual(regs.d["analyzer_trigger_enable"].writes, [0, 1, 0, 1]) + self.assertEqual(regs.d["analyzer_storage_enable"].writes, [0, 1, 0, 1]) + + def test_multi_group_signal_offsets(self): + # A signal present in several groups can sit at different positions in each; triggers + # must resolve it in the currently selected group. + self.tmpdir = tempfile.TemporaryDirectory() + config_csv = os.path.join(self.tmpdir.name, "analyzer.csv") + with open(config_csv, "w") as f: + f.write("config,None,data_width,8\n") + f.write("config,None,depth,16\n") + f.write("config,None,samplerate,100000000\n") + f.write("signal,0,flag,1\n") + f.write("signal,0,shared,3\n") + f.write("signal,1,shared,3\n") + regs = make_regs() + driver = LiteScopeAnalyzerDriver(regs, "analyzer", config_csv=config_csv) + + # Group 0: shared sits above flag; group 1: shared sits at bit 0. + driver.configure_group(0) + driver.add_trigger(cond={"shared": "0b101"}) + self.assertEqual(driver.trigger_terms[-1], (0xe, 0xa)) + + driver.configure_group(1) + driver.add_trigger(cond={"shared": "0b101"}) + self.assertEqual(driver.trigger_terms[-1], (0x7, 0x5)) + def test_configure_subsampler_rejects_invalid_values(self): driver, regs = self.make_driver(subsampler_width=4) From 992d23187081ec71b1e5e02dfdee9cf31f67e263 Mon Sep 17 00:00:00 2001 From: Florent Kermarrec Date: Tue, 14 Jul 2026 09:52:23 +0200 Subject: [PATCH 4/4] test: exercise wide triggers through the CSR bus At data_width > 32 the trigger Mask/Value CSRs are compound (several bus words), but all existing tests poked CSRStorage.write() which sets the whole storage atomically - neither the word-by-word bus path that hosts actually perform nor a comparator wider than 32 bits was ever exercised. Drive a 128-bit-probe analyzer exclusively through a csr_bus.CSRBank, mirroring the host access pattern (MSB word first, ascending addresses, separate arm strobe): check the capture is a consecutive wide pattern, the trigger sample sits at its locked position within the capture, and a second re-armed capture behaves identically. Note: csr_bus.Interface.read samples dat_r one cycle too early for the bank's registered read path; the test uses a local read helper with the settle cycle. --- test/test_analyzer.py | 148 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) diff --git a/test/test_analyzer.py b/test/test_analyzer.py index 8b36669..5077961 100644 --- a/test/test_analyzer.py +++ b/test/test_analyzer.py @@ -4,11 +4,14 @@ # Copyright (c) 2017-2026 Florent Kermarrec # SPDX-License-Identifier: BSD-2-Clause +import re import unittest import tempfile from migen import * +from litex.soc.interconnect import csr_bus + from litescope import LiteScopeAnalyzer from litescope.software.dump.common import DumpData @@ -368,3 +371,148 @@ def get_name(self, signal): self.assertIn("enum,0,fsm_state,0,IDLE", lines) self.assertIn("enum,0,fsm_state,1,RUN", lines) self.assertIn("enum,0,fsm_state,2,DONE", lines) + + +# Wide trigger through the real multi-word CSR bus path --------------------------------------------- +# +# At data_width > 32 the trigger Mask/Value CSRs are compound (multiple bus words). The tests +# above poke CSRStorage.write() which sets the whole storage atomically, so the word-by-word bus +# path (what csr_builder/RemoteClient perform on hardware) was never exercised, and neither was +# a comparator wider than 32 bits. These tests drive the analyzer exclusively through a +# csr_bus.CSRBank, mirroring the host access pattern (MSB word first, ascending addresses, +# separate arm strobe). + +# The trigger sample's position within the capture (offset=8, subsampling=1) is locked below: +# it must stay stable across trigger-pipeline changes (only absolute capture time may shift). + + +class _WideDUT(Module): + def __init__(self, data_width=128, depth=64): + self.counter = counter = Signal(32) + self.sync += counter.eq(counter + 1) + probe = Signal(data_width) + self.comb += probe.eq(Cat(counter, ~counter, (counter + 0x12345678)[:32], (counter ^ 0x55AA55AA)[:32])) + self.submodules.analyzer = LiteScopeAnalyzer(probe, depth, csr_csv=None) + self.bus = csr_bus.Interface(data_width=32, address_width=14) + self.submodules.bank = csr_bus.CSRBank(self.analyzer.get_csrs(), address=0, bus=self.bus) + + +def _wide_pattern(c): + mask32 = 0xffffffff + return (( c & mask32) | + ((~c & mask32) << 32) | + (((c + 0x12345678) & mask32) << 64) | + (((c ^ 0x55AA55AA) & mask32) << 96)) + + +def _csr_addrs(dut, name): + # Bus word addresses of a (possibly compound) CSR; ascending = MSB word first with the + # default "big" ordering, matching litex's csr_builder host-side decomposition. + return [i for i, c in enumerate(dut.bank.simple_csrs) + if re.fullmatch(re.escape(name) + r"\d*", c.name)] + + +def _bus_csr_write(dut, name, value): + addrs = _csr_addrs(dut, name) + n = len(addrs) + for j, adr in enumerate(addrs): + yield from dut.bus.write(adr, (value >> (32*(n - 1 - j))) & 0xffffffff) + + +def _bus_word_read(dut, adr): + # csr_bus.Interface.read samples dat_r one cycle too early for the bank's registered + # read path; add the settle cycle here. + yield dut.bus.adr.eq(adr) + yield dut.bus.re.eq(1) + yield + yield dut.bus.re.eq(0) + yield + return (yield dut.bus.dat_r) + + +def _bus_csr_read(dut, name): + addrs = _csr_addrs(dut, name) + value = 0 + for adr in addrs: + value = (value << 32) | (yield from _bus_word_read(dut, adr)) + return value + + +def _bus_read_samples(dut, count, subwords): + samples = [] + for _ in range(count): + value = 0 + for j in range(subwords): + sub = (yield from _bus_csr_read(dut, "storage_mem_data")) + value |= sub << (32*j) + samples.append(value) + return samples + + +class TestAnalyzerWideTrigger(unittest.TestCase): + # Locked trigger sample position within the capture for offset=8: the matching sample + # lands at index 6. + EXPECTED_TRIGGER_INDEX = 6 + + def _capture(self, dut, trigger_value, offset=8, length=32, timeout=4096): + full_mask = 2**128 - 1 + yield from _bus_csr_write(dut, "subsampler_value", 0) + yield from _bus_csr_write(dut, "storage_offset", offset) + yield from _bus_csr_write(dut, "storage_length", length) + yield from _bus_csr_write(dut, "storage_enable", 0) + yield from _bus_csr_write(dut, "storage_enable", 1) + yield from _bus_csr_write(dut, "trigger_enable", 0) + yield from _bus_csr_write(dut, "trigger_mem_mask", full_mask) + yield from _bus_csr_write(dut, "trigger_mem_value", trigger_value) + yield from _bus_csr_write(dut, "trigger_mem_write", 1) + yield from _bus_csr_write(dut, "trigger_enable", 1) + # done stays asserted (IDLE) until the enable edge crosses into the scope domain: + # wait for the capture to start before waiting for it to complete. + for _ in range(timeout): + if not (yield from _bus_csr_read(dut, "storage_done")): + break + yield + else: + self.fail("capture did not start") + for _ in range(timeout): + if (yield from _bus_csr_read(dut, "storage_done")): + break + yield + else: + self.fail("capture did not complete") + # After completion the storage FIFO starts migrating samples into the CDC/read + # pipeline, so mem_level only reflects what still sits in the FIFO; the read path + # delivers the full capture. + level = yield from _bus_csr_read(dut, "storage_mem_level") + self.assertGreater(level, 0) + return (yield from _bus_read_samples(dut, length, 4)) + + def test_wide_trigger_via_csr_bus(self): + dut = _WideDUT() + results = {} + + def generator(): + yield + targets = [] + for run in range(2): + # Re-arm on each iteration: terms are consumed per capture and must be + # re-loaded (what the driver's run() does). + target = (yield dut.counter) + 500 + targets.append(target) + results[run] = (yield from self._capture(dut, _wide_pattern(target))) + results["targets"] = targets + + run_simulation(dut, {"sys": [generator()]}, {"sys": 10, "scope": 10}, vcd_name=None) + + for run in range(2): + samples = results[run] + base = samples[0] & 0xffffffff + self.assertEqual(samples, [_wide_pattern(base + i) for i in range(len(samples))], + msg=f"capture {run} is not a consecutive wide pattern") + # The trigger sample sits at the locked position within the capture. + self.assertEqual(samples[self.EXPECTED_TRIGGER_INDEX], + _wide_pattern(results["targets"][run])) + + +if __name__ == "__main__": + unittest.main()