usb-tools/amaranth
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

back.pysim: redesign the simulator.

Browse source 
The redesign introduces no fundamental incompatibilities, but it does
involve minor breaking changes:
  * The simulator commands were moved from hdl.ast to back.pysim
    (instead of only being reexported from back.pysim).
  * back.pysim.DeadlineError was removed.

Summary of changes:
  * The new simulator compiles HDL to Python code and is >6x faster.
    (The old one compiled HDL to lots of Python lambdas.)
  * The new simulator is a straightforward, rigorous implementation
    of the Synchronous Reactive Programming paradigm, instead of
    a pile of ad-hoc code with no particular design driving it.
  * The new simulator never raises DeadlineError, and there is no
    limit on the amount of delta cycles.
  * The new simulator robustly handles multiclock designs.
  * The new simulator can be reset, such that the compiled design
    can be reused, which can save significant runtime with large
    designs.
  * Generators can no longer be added as processes, since that would
    break reset(); only generator functions may be. If necessary,
    they may be added by wrapping them into a generator function;
    a deprecated fallback does just that. This workaround will raise
    an exception if the simulator is reset and restarted.
  * The new simulator does not depend on Python extensions.
    (The old one required bitarray, which did not provide wheels.)

Fixes #28.
Fixes #34.
Fixes #160.
Fixes #161.
Fixes #215.
Fixes #242.
Fixes #262.
This commit is contained in:
whitequark 2019-11-22 08:32:41 +00:00
parent f8428ff505
commit 7df70059d1
7 changed files with 1164 additions and 906 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
examples/basic/ctr_en.py
View file

@ -19,17 +19,15 @@ ctr = Counter(width=16)
print(verilog.convert(ctr, ports=[ctr.o, ctr.en]))
with pysim.Simulator(ctr,
vcd_file=open("ctrl.vcd", "w"),
gtkw_file=open("ctrl.gtkw", "w"),
traces=[ctr.en, ctr.v, ctr.o]) as sim:
sim.add_clock(1e-6)
def ce_proc():
yield; yield; yield
yield ctr.en.eq(1)
yield; yield; yield
yield ctr.en.eq(0)
yield; yield; yield
yield ctr.en.eq(1)
sim.add_sync_process(ce_proc())
sim = pysim.Simulator(ctr)
sim.add_clock(1e-6)
def ce_proc():
yield; yield; yield
yield ctr.en.eq(1)
yield; yield; yield
yield ctr.en.eq(0)
yield; yield; yield
yield ctr.en.eq(1)
sim.add_sync_process(ce_proc)
with sim.write_vcd("ctrl.vcd", "ctrl.gtkw", traces=[ctr.en, ctr.v, ctr.o]):
sim.run_until(100e-6, run_passive=True)

1670
nmigen/back/pysim.py
View file

File diff suppressed because it is too large Load diff

27
nmigen/compat/sim/__init__.py
View file

@ -21,15 +21,24 @@ def run_simulation(fragment_or_module, generators, clocks={"sync": 10}, vcd_name
generators = {"sync": generators}
fragment.domains += ClockDomain("sync")
with Simulator(fragment, vcd_file=open(vcd_name, "w") if vcd_name else None) as sim:
for domain, period in clocks.items():
sim.add_clock(period / 1e9, domain=domain)
for domain, processes in generators.items():
if isinstance(processes, Iterable) and not inspect.isgenerator(processes):
for process in processes:
sim.add_sync_process(process, domain=domain)
else:
sim.add_sync_process(processes, domain=domain)
sim = Simulator(fragment)
for domain, period in clocks.items():
sim.add_clock(period / 1e9, domain=domain)
for domain, processes in generators.items():
def wrap(process):
def wrapper():
yield from process
return wrapper
if isinstance(processes, Iterable) and not inspect.isgenerator(processes):
for process in processes:
sim.add_sync_process(wrap(process), domain=domain)
else:
sim.add_sync_process(wrap(processes), domain=domain)
if vcd_name is not None:
with sim.write_vcd(vcd_name):
sim.run()
else:
sim.run()

107
nmigen/test/test_lib_cdc.py
View file

@ -19,37 +19,39 @@ class FFSynchronizerTestCase(FHDLTestCase):
i = Signal()
o = Signal()
frag = FFSynchronizer(i, o)
with Simulator(frag) as sim:
sim.add_clock(1e-6)
def process():
self.assertEqual((yield o), 0)
yield i.eq(1)
yield Tick()
self.assertEqual((yield o), 0)
yield Tick()
self.assertEqual((yield o), 0)
yield Tick()
self.assertEqual((yield o), 1)
sim.add_process(process)
sim.run()
sim = Simulator(frag)
sim.add_clock(1e-6)
def process():
self.assertEqual((yield o), 0)
yield i.eq(1)
yield Tick()
self.assertEqual((yield o), 0)
yield Tick()
self.assertEqual((yield o), 0)
yield Tick()
self.assertEqual((yield o), 1)
sim.add_process(process)
sim.run()
def test_reset_value(self):
i = Signal(reset=1)
o = Signal()
frag = FFSynchronizer(i, o, reset=1)
with Simulator(frag) as sim:
sim.add_clock(1e-6)
def process():
self.assertEqual((yield o), 1)
yield i.eq(0)
yield Tick()
self.assertEqual((yield o), 1)
yield Tick()
self.assertEqual((yield o), 1)
yield Tick()
self.assertEqual((yield o), 0)
sim.add_process(process)
sim.run()
sim = Simulator(frag)
sim.add_clock(1e-6)
def process():
self.assertEqual((yield o), 1)
yield i.eq(0)
yield Tick()
self.assertEqual((yield o), 1)
yield Tick()
self.assertEqual((yield o), 1)
yield Tick()
self.assertEqual((yield o), 0)
sim.add_process(process)
sim.run()
class ResetSynchronizerTestCase(FHDLTestCase):
@ -69,31 +71,32 @@ class ResetSynchronizerTestCase(FHDLTestCase):
s = Signal(reset=1)
m.d.sync += s.eq(0)
with Simulator(m, vcd_file=open("test.vcd", "w")) as sim:
sim.add_clock(1e-6)
def process():
# initial reset
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 0)
yield Tick(); yield Delay(1e-8)
sim = Simulator(m)
sim.add_clock(1e-6)
def process():
# initial reset
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 0)
yield Tick(); yield Delay(1e-8)
yield arst.eq(1)
yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield arst.eq(0)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 0)
yield Tick(); yield Delay(1e-8)
sim.add_process(process)
yield arst.eq(1)
yield Delay(1e-8)
self.assertEqual((yield s), 0)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield arst.eq(0)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 1)
yield Tick(); yield Delay(1e-8)
self.assertEqual((yield s), 0)
yield Tick(); yield Delay(1e-8)
sim.add_process(process)
with sim.write_vcd("test.vcd"):
sim.run()

100
nmigen/test/test_lib_coding.py
View file

@ -8,78 +8,78 @@ from ..lib.coding import *
class EncoderTestCase(FHDLTestCase):
def test_basic(self):
enc = Encoder(4)
with Simulator(enc) as sim:
def process():
self.assertEqual((yield enc.n), 1)
self.assertEqual((yield enc.o), 0)
def process():
self.assertEqual((yield enc.n), 1)
self.assertEqual((yield enc.o), 0)
yield enc.i.eq(0b0001)
yield Delay()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 0)
yield enc.i.eq(0b0001)
yield Settle()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 0)
yield enc.i.eq(0b0100)
yield Delay()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 2)
yield enc.i.eq(0b0100)
yield Settle()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 2)
yield enc.i.eq(0b0110)
yield Delay()
self.assertEqual((yield enc.n), 1)
self.assertEqual((yield enc.o), 0)
yield enc.i.eq(0b0110)
yield Settle()
self.assertEqual((yield enc.n), 1)
self.assertEqual((yield enc.o), 0)
sim.add_process(process)
sim.run()
sim = Simulator(enc)
sim.add_process(process)
sim.run()
class PriorityEncoderTestCase(FHDLTestCase):
def test_basic(self):
enc = PriorityEncoder(4)
with Simulator(enc) as sim:
def process():
self.assertEqual((yield enc.n), 1)
self.assertEqual((yield enc.o), 0)
def process():
self.assertEqual((yield enc.n), 1)
self.assertEqual((yield enc.o), 0)
yield enc.i.eq(0b0001)
yield Delay()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 0)
yield enc.i.eq(0b0001)
yield Settle()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 0)
yield enc.i.eq(0b0100)
yield Delay()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 2)
yield enc.i.eq(0b0100)
yield Settle()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 2)
yield enc.i.eq(0b0110)
yield Delay()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 1)
yield enc.i.eq(0b0110)
yield Settle()
self.assertEqual((yield enc.n), 0)
self.assertEqual((yield enc.o), 1)
sim.add_process(process)
sim.run()
sim = Simulator(enc)
sim.add_process(process)
sim.run()
class DecoderTestCase(FHDLTestCase):
def test_basic(self):
dec = Decoder(4)
with Simulator(dec) as sim:
def process():
self.assertEqual((yield dec.o), 0b0001)
def process():
self.assertEqual((yield dec.o), 0b0001)
yield dec.i.eq(1)
yield Delay()
self.assertEqual((yield dec.o), 0b0010)
yield dec.i.eq(1)
yield Settle()
self.assertEqual((yield dec.o), 0b0010)
yield dec.i.eq(3)
yield Delay()
self.assertEqual((yield dec.o), 0b1000)
yield dec.i.eq(3)
yield Settle()
self.assertEqual((yield dec.o), 0b1000)
yield dec.n.eq(1)
yield Delay()
self.assertEqual((yield dec.o), 0b0000)
yield dec.n.eq(1)
yield Settle()
self.assertEqual((yield dec.o), 0b0000)
sim.add_process(process)
sim.run()
sim = Simulator(dec)
sim.add_process(process)
sim.run()
class ReversibleSpec(Elaboratable):

136
nmigen/test/test_sim.py
View file

@ -1,3 +1,4 @@
import os
from contextlib import contextmanager
from .utils import *
@ -25,16 +26,14 @@ class SimulatorUnitTestCase(FHDLTestCase):
for signal in flatten(s._lhs_signals() for s in Statement.cast(stmt)):
frag.add_driver(signal)
with Simulator(frag,
vcd_file =open("test.vcd", "w"),
gtkw_file=open("test.gtkw", "w"),
traces=[*isigs, osig]) as sim:
def process():
for isig, input in zip(isigs, inputs):
yield isig.eq(input)
yield Delay()
self.assertEqual((yield osig), output.value)
sim.add_process(process)
sim = Simulator(frag)
def process():
for isig, input in zip(isigs, inputs):
yield isig.eq(input)
yield Settle()
self.assertEqual((yield osig), output.value)
sim.add_process(process)
with sim.write_vcd("test.vcd", "test.gtkw", traces=[*isigs, osig]):
sim.run()
def test_invert(self):
@ -213,6 +212,13 @@ class SimulatorUnitTestCase(FHDLTestCase):
stmt = lambda y, a: [Cat(l, m, n).eq(a), y.eq(Cat(n, m, l))]
self.assertStatement(stmt, [C(0b100101110, 9)], C(0b110101100, 9))
def test_nested_cat_lhs(self):
l = Signal(3)
m = Signal(3)
n = Signal(3)
stmt = lambda y, a: [Cat(Cat(l, Cat(m)), n).eq(a), y.eq(Cat(n, m, l))]
self.assertStatement(stmt, [C(0b100101110, 9)], C(0b110101100, 9))
def test_record(self):
rec = Record([
("l", 1),
@ -277,8 +283,9 @@ class SimulatorUnitTestCase(FHDLTestCase):
class SimulatorIntegrationTestCase(FHDLTestCase):
@contextmanager
def assertSimulation(self, module, deadline=None):
with Simulator(module) as sim:
yield sim
sim = Simulator(module)
yield sim
with sim.write_vcd("test.vcd", "test.gtkw"):
if deadline is None:
sim.run()
else:
@ -300,11 +307,15 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
yield Delay(1e-6)
self.assertEqual((yield self.count), 4)
yield self.sync.clk.eq(1)
self.assertEqual((yield self.count), 4)
yield Settle()
self.assertEqual((yield self.count), 5)
yield Delay(1e-6)
self.assertEqual((yield self.count), 5)
yield self.sync.clk.eq(0)
self.assertEqual((yield self.count), 5)
yield Settle()
self.assertEqual((yield self.count), 5)
for _ in range(3):
yield Delay(1e-6)
yield self.sync.clk.eq(1)
@ -328,6 +339,26 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
self.assertEqual((yield self.count), 0)
sim.add_sync_process(process)
def test_reset(self):
self.setUp_counter()
sim = Simulator(self.m)
sim.add_clock(1e-6)
times = 0
def process():
nonlocal times
self.assertEqual((yield self.count), 4)
yield
self.assertEqual((yield self.count), 5)
yield
self.assertEqual((yield self.count), 6)
yield
times += 1
sim.add_sync_process(process)
sim.run()
sim.reset()
sim.run()
self.assertEqual(times, 2)
def setUp_alu(self):
self.a = Signal(8)
self.b = Signal(8)
@ -406,7 +437,7 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
def process():
yield self.i.eq(0b10101010)
yield self.i[:4].eq(-1)
yield Delay()
yield Settle()
self.assertEqual((yield self.i[:4]), 0b1111)
self.assertEqual((yield self.i), 0b10101111)
sim.add_process(process)
@ -426,10 +457,18 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
def test_add_process_wrong(self):
with self.assertSimulation(Module()) as sim:
with self.assertRaises(TypeError,
msg="Cannot add a process 1 because it is not a generator or "
"a generator function"):
msg="Cannot add a process 1 because it is not a generator function"):
sim.add_process(1)
def test_add_process_wrong_generator(self):
with self.assertSimulation(Module()) as sim:
with self.assertWarns(DeprecationWarning,
msg="instead of generators, use generator functions as processes; "
"this allows the simulator to be repeatedly reset"):
def process():
yield Delay()
sim.add_process(process())
def test_add_clock_wrong_twice(self):
m = Module()
s = Signal()
@ -452,37 +491,18 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
with self.assertSimulation(m) as sim:
sim.add_clock(1, if_exists=True)
def test_eq_signal_unused_wrong(self):
self.setUp_lhs_rhs()
self.s = Signal()
with self.assertSimulation(self.m) as sim:
def process():
with self.assertRaisesRegex(ValueError,
regex=r"Process .+? sent a request to set signal \(sig s\), "
r"which is not a part of simulation"):
yield self.s.eq(0)
yield Delay()
sim.add_process(process)
def test_eq_signal_comb_wrong(self):
self.setUp_lhs_rhs()
with self.assertSimulation(self.m) as sim:
def process():
with self.assertRaisesRegex(ValueError,
regex=r"Process .+? sent a request to set signal \(sig o\), "
r"which is a part of combinatorial assignment in simulation"):
yield self.o.eq(0)
yield Delay()
sim.add_process(process)
def test_command_wrong(self):
survived = False
with self.assertSimulation(Module()) as sim:
def process():
nonlocal survived
with self.assertRaisesRegex(TypeError,
regex=r"Received unsupported command 1 from process .+?"):
yield 1
yield Delay()
yield Settle()
survived = True
sim.add_process(process)
self.assertTrue(survived)
def setUp_memory(self, rd_synchronous=True, rd_transparent=True, wr_granularity=None):
self.m = Module()
@ -558,7 +578,7 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
self.assertEqual((yield self.rdport.data), 0xaa)
yield
self.assertEqual((yield self.rdport.data), 0xaa)
yield Delay(1e-6) # let comb propagate
yield Settle()
self.assertEqual((yield self.rdport.data), 0x33)
sim.add_clock(1e-6)
sim.add_sync_process(process)
@ -571,11 +591,11 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
yield self.wrport.en.eq(1)
yield
self.assertEqual((yield self.rdport.data), 0xaa)
yield Delay(1e-6) # let comb propagate
yield Settle()
self.assertEqual((yield self.rdport.data), 0x33)
yield
yield self.rdport.addr.eq(1)
yield Delay(1e-6) # let comb propagate
yield Settle()
self.assertEqual((yield self.rdport.data), 0x33)
sim.add_clock(1e-6)
sim.add_sync_process(process)
@ -585,10 +605,10 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
with self.assertSimulation(self.m) as sim:
def process():
yield self.rdport.addr.eq(0)
yield Delay()
yield Settle()
self.assertEqual((yield self.rdport.data), 0xaa)
yield self.rdport.addr.eq(1)
yield Delay()
yield Settle()
self.assertEqual((yield self.rdport.data), 0x55)
yield self.rdport.addr.eq(0)
yield self.wrport.addr.eq(0)
@ -596,7 +616,7 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
yield self.wrport.en.eq(1)
yield Tick("sync")
self.assertEqual((yield self.rdport.data), 0xaa)
yield Delay(1e-6) # let comb propagate
yield Settle()
self.assertEqual((yield self.rdport.data), 0x33)
sim.add_clock(1e-6)
sim.add_process(process)
@ -661,8 +681,26 @@ class SimulatorIntegrationTestCase(FHDLTestCase):
sim.add_sync_process(process_gen)
sim.add_sync_process(process_check)
def test_wrong_not_run(self):
with self.assertWarns(UserWarning,
msg="Simulation created, but not run"):
with Simulator(Fragment()) as sim:
def test_vcd_wrong_nonzero_time(self):
s = Signal()
m = Module()
m.d.sync += s.eq(s)
sim = Simulator(m)
sim.add_clock(1e-6)
sim.run_until(1e-5)
with self.assertRaisesRegex(ValueError,
regex=r"^Cannot start writing waveforms after advancing simulation time$"):
with sim.write_vcd(open(os.path.devnull, "wt")):
pass
def test_vcd_wrong_twice(self):
s = Signal()
m = Module()
m.d.sync += s.eq(s)
sim = Simulator(m)
sim.add_clock(1e-6)
with self.assertRaisesRegex(ValueError,
regex=r"^Already writing waveforms to .+$"):
with sim.write_vcd(open(os.path.devnull, "wt")):
with sim.write_vcd(open(os.path.devnull, "wt")):
pass

6
setup.py
View file

@ -24,7 +24,11 @@ setup(
license="BSD",
python_requires="~=3.6",
setup_requires=["setuptools_scm"],
install_requires=["setuptools", "pyvcd>=0.1.4", "bitarray", "Jinja2"],
install_requires=[
"setuptools",
"pyvcd~=0.1.4", # for nmigen.pysim
"Jinja2", # for nmigen.build
],
packages=find_packages(),
entry_points={
"console_scripts": [