from .. import *
from ..formal import *
+from ..tools import log2_int
+from .coding import GrayEncoder
-__all__ = ["FIFOInterface", "SyncFIFO", "SyncFIFOBuffered"]
+__all__ = ["FIFOInterface", "SyncFIFO", "SyncFIFOBuffered", "AsyncFIFO", "AsyncFIFOBuffered"]
class FIFOInterface:
Attributes
----------
+ {attributes}
din : in, width
Input data.
writable : out
""",
parameters="",
dout_valid="The conditions in which ``dout`` is valid depends on the type of the queue.",
+ attributes="""
+ fwft : bool
+ First-word fallthrough. If set, when ``readable`` rises, the first entry is already
+ available, i.e. ``dout`` is valid. Otherwise, after ``readable`` rises, it is necessary
+ to strobe ``re`` for ``dout`` to become valid.
+ """.strip(),
w_attributes="",
r_attributes="")
- def __init__(self, width, depth):
+ def __init__(self, width, depth, fwft):
self.width = width
self.depth = depth
+ self.fwft = fwft
self.din = Signal(width, reset_less=True)
self.writable = Signal() # not full
For FWFT queues, valid if ``readable`` is asserted. For non-FWFT queues, valid on the next
cycle after ``readable`` and ``re`` have been asserted.
""".strip(),
+ attributes="",
r_attributes="""
level : out
Number of unread entries.
""".strip())
def __init__(self, width, depth, fwft=True):
- super().__init__(width, depth)
-
- self.fwft = fwft
+ super().__init__(width, depth, fwft)
self.level = Signal(max=depth + 1)
self.replace = Signal()
class SyncFIFOBuffered(FIFOInterface):
- """
+ __doc__ = FIFOInterface._doc_template.format(
+ description="""
Buffered synchronous first in, first out queue.
This queue's interface is identical to :class:`SyncFIFO` configured as ``fwft=True``, but it
In exchange, the latency between an entry being written to an empty queue and that entry
becoming available on the output is increased to one cycle.
- """
- def __init__(self, width, depth):
- super().__init__(width, depth)
+ """.strip(),
+ parameters="""
+ fwft : bool
+ Always set.
+ """.strip(),
+ attributes="",
+ dout_valid="Valid if ``readable`` is asserted.",
+ r_attributes="""
+ level : out
+ Number of unread entries.
+ """.strip(),
+ w_attributes="")
- self.fwft = True
+ def __init__(self, width, depth):
+ super().__init__(width, depth, fwft=True)
self.level = Signal(max=depth + 1)
m.d.comb += self.level.eq(fifo.level + self.readable)
return m.lower(platform)
+
+
+class AsyncFIFO(FIFOInterface):
+ __doc__ = FIFOInterface._doc_template.format(
+ description="""
+ Asynchronous first in, first out queue.
+
+ Read and write interfaces are accessed from different clock domains, called ``read``
+ and ``write``; use :class:`ClockDomainsRenamer` to rename them as appropriate for the design.
+ """.strip(),
+ parameters="""
+ fwft : bool
+ Always set.
+ """.strip(),
+ attributes="",
+ dout_valid="Valid if ``readable`` is asserted.",
+ r_attributes="",
+ w_attributes="")
+
+ def __init__(self, width, depth):
+ super().__init__(width, depth, fwft=True)
+
+ try:
+ self._ctr_bits = log2_int(depth, need_pow2=True) + 1
+ except ValueError as e:
+ raise ValueError("AsyncFIFO only supports power-of-2 depths") from e
+
+ def get_fragment(self, platform):
+ # The design of this queue is the "style #2" from Clifford E. Cummings' paper "Simulation
+ # and Synthesis Techniques for Asynchronous FIFO Design":
+ # http://www.sunburst-design.com/papers/CummingsSNUG2002SJ_FIFO1.pdf
+
+ m = Module()
+
+ produce_w_bin = Signal(self._ctr_bits)
+ produce_w_gry = Signal(self._ctr_bits)
+ produce_r_gry = Signal(self._ctr_bits)
+ produce_enc = m.submodules.produce_enc = \
+ GrayEncoder(self._ctr_bits)
+ produce_cdc = m.submodules.produce_cdc = \
+ MultiReg(produce_w_gry, produce_r_gry, odomain="read")
+ m.d.comb += [
+ produce_enc.i.eq(produce_w_bin),
+ produce_w_gry.eq(produce_enc.o),
+ ]
+
+ consume_r_bin = Signal(self._ctr_bits)
+ consume_r_gry = Signal(self._ctr_bits)
+ consume_w_gry = Signal(self._ctr_bits)
+ consume_enc = m.submodules.consume_enc = \
+ GrayEncoder(self._ctr_bits)
+ consume_cdc = m.submodules.consume_cdc = \
+ MultiReg(consume_r_gry, consume_w_gry, odomain="write")
+ m.d.comb += [
+ consume_enc.i.eq(consume_r_bin),
+ consume_r_gry.eq(consume_enc.o),
+ ]
+
+ m.d.comb += [
+ self.writable.eq(
+ (produce_w_gry[-1] == consume_w_gry[-1]) |
+ (produce_w_gry[-2] == consume_w_gry[-2]) |
+ (produce_w_gry[:-2] != consume_w_gry[:-2])),
+ self.readable.eq(consume_r_gry != produce_r_gry)
+ ]
+
+ do_write = self.writable & self.we
+ do_read = self.readable & self.re
+ m.d.write += produce_w_bin.eq(produce_w_bin + do_write)
+ m.d.read += consume_r_bin.eq(consume_r_bin + do_read)
+
+ storage = Memory(self.width, self.depth)
+ wrport = m.submodules.wrport = storage.write_port(domain="write")
+ rdport = m.submodules.rdport = storage.read_port (domain="read")
+ m.d.comb += [
+ wrport.addr.eq(produce_w_bin[:-1]),
+ wrport.data.eq(self.din),
+ wrport.en.eq(do_write)
+ ]
+ m.d.comb += [
+ rdport.addr.eq((consume_r_bin + do_read)[:-1]),
+ self.dout.eq(rdport.data),
+ ]
+
+ return m.lower(platform)
+
+
+class AsyncFIFOBuffered(FIFOInterface):
+ __doc__ = FIFOInterface._doc_template.format(
+ description="""
+ Buffered asynchronous first in, first out queue.
+
+ This queue's interface is identical to :class:`AsyncFIFO`, but it has an additional register
+ on the output, improving timing in case of block RAM that has large clock-to-output delay.
+
+ In exchange, the latency between an entry being written to an empty queue and that entry
+ becoming available on the output is increased to one cycle.
+ """.strip(),
+ parameters="""
+ fwft : bool
+ Always set.
+ """.strip(),
+ attributes="",
+ dout_valid="Valid if ``readable`` is asserted.",
+ r_attributes="",
+ w_attributes="")
+
+ def __init__(self, width, depth):
+ super().__init__(width, depth, fwft=True)
+
+ def get_fragment(self, platform):
+ m = Module()
+ m.submodules.unbuffered = fifo = AsyncFIFO(self.width, self.depth - 1)
+
+ m.d.comb += [
+ fifo.din.eq(self.din),
+ self.writable.eq(fifo.writable),
+ fifo.we.eq(self.we),
+ ]
+
+ with m.If(self.re | ~self.readable):
+ m.d.read += [
+ self.dout.eq(fifo.dout),
+ self.readable.eq(fifo.readable)
+ ]
+ m.d.comb += \
+ fifo.re.eq(1)
+
+ return m.lower(platform)
from ..hdl.dsl import *
from ..hdl.mem import *
from ..hdl.ir import *
+from ..hdl.xfrm import *
+from ..hdl.cd import *
from ..back.pysim import *
from ..lib.fifo import *
class FIFOSmokeTestCase(FHDLTestCase):
- def assertSyncFIFOWorks(self, fifo):
- with Simulator(fifo) as sim:
+ def assertSyncFIFOWorks(self, fifo, xfrm=lambda x: x):
+ with Simulator(xfrm(fifo.get_fragment(None)), vcd_file=open("test.vcd", "w")) as sim:
sim.add_clock(1e-6)
def process():
yield from fifo.write(1)
yield from fifo.write(2)
+ while not (yield fifo.readable):
+ yield
+ if not fifo.fwft:
+ yield fifo.re.eq(1)
yield
self.assertEqual((yield from fifo.read()), 1)
self.assertEqual((yield from fifo.read()), 2)
sim.add_sync_process(process)
sim.run()
+ def assertAsyncFIFOWorks(self, fifo):
+ self.assertSyncFIFOWorks(fifo, xfrm=DomainRenamer({"read": "sync", "write": "sync"}))
+
def test_sync_fwft(self):
- fifo = SyncFIFO(width=8, depth=4, fwft=True)
- self.assertSyncFIFOWorks(SyncFIFO(width=8, depth=4))
+ self.assertSyncFIFOWorks(SyncFIFO(width=8, depth=4, fwft=True))
def test_sync_not_fwft(self):
- fifo = SyncFIFO(width=8, depth=4, fwft=False)
- self.assertSyncFIFOWorks(SyncFIFO(width=8, depth=4))
+ self.assertSyncFIFOWorks(SyncFIFO(width=8, depth=4, fwft=False))
def test_sync_buffered(self):
- fifo = SyncFIFO(width=8, depth=4, fwft=True)
- self.assertSyncFIFOWorks(SyncFIFOBuffered(width=8, depth=4))
+ self.assertSyncFIFOWorks(SyncFIFO(width=8, depth=4, fwft=True))
+
+ def test_async(self):
+ self.assertAsyncFIFOWorks(AsyncFIFO(width=8, depth=4))
+
+ def test_async_buffered(self):
+ self.assertAsyncFIFOWorks(AsyncFIFOBuffered(width=8, depth=3))
class FIFOModel(FIFOInterface):
"""
Non-synthesizable first-in first-out queue, implemented naively as a chain of registers.
"""
- def __init__(self, width, depth, fwft):
- super().__init__(width, depth)
+ def __init__(self, width, depth, fwft, rdomain, wdomain):
+ super().__init__(width, depth, fwft)
- self.fwft = fwft
+ self.rdomain = rdomain
+ self.wdomain = wdomain
self.replace = Signal()
self.level = Signal(max=self.depth + 1)
m = Module()
storage = Memory(self.width, self.depth)
- wrport = m.submodules.wrport = storage.write_port()
- rdport = m.submodules.rdport = storage.read_port(synchronous=False)
+ wrport = m.submodules.wrport = storage.write_port(domain=self.wdomain)
+ rdport = m.submodules.rdport = storage.read_port (synchronous=False)
produce = Signal(max=self.depth)
consume = Signal(max=self.depth)
m.d.comb += self.dout.eq(rdport.data)
with m.If(self.re & self.readable):
if not self.fwft:
- m.d.sync += self.dout.eq(rdport.data)
- m.d.sync += consume.eq(rdport.addr)
+ m.d[self.rdomain] += self.dout.eq(rdport.data)
+ m.d[self.rdomain] += consume.eq(rdport.addr)
m.d.comb += self.writable.eq(self.level < self.depth)
m.d.comb += wrport.data.eq(self.din)
with m.If(~self.replace & self.writable):
m.d.comb += wrport.addr.eq((produce + 1) % self.depth)
m.d.comb += wrport.en.eq(1)
- m.d.sync += produce.eq(wrport.addr)
+ m.d[self.wdomain] += produce.eq(wrport.addr)
with m.If(self.replace):
# The result of trying to replace an element in an empty queue is irrelevant.
# The result of trying to replace the element that is currently being read
m.d.comb += wrport.addr.eq(produce)
m.d.comb += wrport.en.eq(1)
- m.d.sync += self.level.eq(self.level
- + (self.writable & self.we & ~self.replace)
- - (self.readable & self.re))
+ with m.If(ResetSignal(self.rdomain) | ResetSignal(self.wdomain)):
+ m.d.sync += self.level.eq(0)
+ with m.Else():
+ m.d.sync += self.level.eq(self.level
+ + (self.writable & self.we & ~self.replace)
+ - (self.readable & self.re))
+
+ m.d.comb += Assert(ResetSignal(self.rdomain) == ResetSignal(self.wdomain))
return m.lower(platform)
signals, the behavior of the implementation under test exactly matches the ideal model,
except for behavior not defined by the model.
"""
- def __init__(self, fifo):
+ def __init__(self, fifo, rdomain, wdomain):
self.fifo = fifo
+ self.rdomain = rdomain
+ self.wdomain = wdomain
+
def get_fragment(self, platform):
m = Module()
m.submodules.dut = dut = self.fifo
- m.submodules.gold = gold = FIFOModel(dut.width, dut.depth, dut.fwft)
+ m.submodules.gold = gold = FIFOModel(dut.width, dut.depth, dut.fwft,
+ self.rdomain, self.wdomain)
m.d.comb += [
gold.re.eq(dut.readable & dut.re),
m.d.comb += gold.replace.eq(0)
m.d.comb += Assert(dut.readable.implies(gold.readable))
+ m.d.comb += Assert(dut.writable.implies(gold.writable))
+ if hasattr(dut, "level"):
+ m.d.comb += Assert(dut.level == gold.level)
+
if dut.fwft:
m.d.comb += Assert(dut.readable
.implies(dut.dout == gold.dout))
else:
- m.d.comb += Assert((Past(dut.readable) & Past(dut.re))
+ m.d.comb += Assert((Past(dut.readable, domain=self.rdomain) &
+ Past(dut.re, domain=self.rdomain))
.implies(dut.dout == gold.dout))
- m.d.comb += Assert(dut.writable == gold.writable)
-
- if hasattr(dut, "level"):
- m.d.comb += Assert(dut.level == gold.level)
-
return m.lower(platform)
consecutively, they must be read out consecutively at some later point, no matter all other
circumstances, with the exception of reset.
"""
- def __init__(self, fifo, bound):
- self.fifo = fifo
- self.bound = bound
+ def __init__(self, fifo, rdomain, wdomain, bound):
+ self.fifo = fifo
+ self.rdomain = rdomain
+ self.wdomain = wdomain
+ self.bound = bound
def get_fragment(self, platform):
m = Module()
m.submodules.dut = fifo = self.fifo
+ m.domains += ClockDomain("sync")
m.d.comb += ResetSignal().eq(0)
+ if self.wdomain != "sync":
+ m.domains += ClockDomain(self.wdomain)
+ m.d.comb += ResetSignal(self.wdomain).eq(0)
+ if self.rdomain != "sync":
+ m.domains += ClockDomain(self.rdomain)
+ m.d.comb += ResetSignal(self.rdomain).eq(0)
+
if hasattr(fifo, "replace"):
m.d.comb += fifo.replace.eq(0)
entry_1 = AnyConst(fifo.width)
entry_2 = AnyConst(fifo.width)
- with m.FSM() as write_fsm:
+ with m.FSM(domain=self.wdomain) as write_fsm:
with m.State("WRITE-1"):
with m.If(fifo.writable):
m.d.comb += [
]
m.next = "DONE"
- with m.FSM() as read_fsm:
+ with m.FSM(domain=self.rdomain) as read_fsm:
read_1 = Signal(fifo.width)
read_2 = Signal(fifo.width)
with m.State("READ"):
m.d.comb += fifo.re.eq(1)
- with m.If(fifo.readable if fifo.fwft else Past(fifo.readable)):
+ if fifo.fwft:
+ readable = fifo.readable
+ else:
+ readable = Past(fifo.readable, domain=self.rdomain)
+ with m.If(readable):
m.d.sync += [
read_1.eq(read_2),
read_2.eq(fifo.dout),
with m.If(Past(initstate, self.bound - 1)):
m.d.comb += Assert(read_fsm.ongoing("DONE"))
+ if self.wdomain != "sync" or self.rdomain != "sync":
+ m.d.comb += Assume(Rose(ClockSignal(self.wdomain)) |
+ Rose(ClockSignal(self.rdomain)))
+
return m.lower(platform)
class FIFOFormalCase(FHDLTestCase):
- def check_fifo(self, fifo):
- self.assertFormal(FIFOModelEquivalenceSpec(fifo),
+ def check_sync_fifo(self, fifo):
+ self.assertFormal(FIFOModelEquivalenceSpec(fifo, rdomain="sync", wdomain="sync"),
mode="bmc", depth=fifo.depth + 1)
- self.assertFormal(FIFOContractSpec(fifo, bound=fifo.depth * 2 + 1),
+ self.assertFormal(FIFOContractSpec(fifo, rdomain="sync", wdomain="sync",
+ bound=fifo.depth * 2 + 1),
mode="hybrid", depth=fifo.depth * 2 + 1)
def test_sync_fwft_pot(self):
- self.check_fifo(SyncFIFO(width=8, depth=4, fwft=True))
+ self.check_sync_fifo(SyncFIFO(width=8, depth=4, fwft=True))
def test_sync_fwft_npot(self):
- self.check_fifo(SyncFIFO(width=8, depth=5, fwft=True))
+ self.check_sync_fifo(SyncFIFO(width=8, depth=5, fwft=True))
def test_sync_not_fwft_pot(self):
- self.check_fifo(SyncFIFO(width=8, depth=4, fwft=False))
+ self.check_sync_fifo(SyncFIFO(width=8, depth=4, fwft=False))
def test_sync_not_fwft_npot(self):
- self.check_fifo(SyncFIFO(width=8, depth=5, fwft=False))
+ self.check_sync_fifo(SyncFIFO(width=8, depth=5, fwft=False))
def test_sync_buffered_pot(self):
- self.check_fifo(SyncFIFOBuffered(width=8, depth=4))
+ self.check_sync_fifo(SyncFIFOBuffered(width=8, depth=4))
def test_sync_buffered_potp1(self):
- self.check_fifo(SyncFIFOBuffered(width=8, depth=5))
+ self.check_sync_fifo(SyncFIFOBuffered(width=8, depth=5))
def test_sync_buffered_potm1(self):
- self.check_fifo(SyncFIFOBuffered(width=8, depth=3))
+ self.check_sync_fifo(SyncFIFOBuffered(width=8, depth=3))
+
+ def check_async_fifo(self, fifo):
+ # TODO: properly doing model equivalence checking on this likely requires multiclock,
+ # which is not really documented nor is it clear how to use it.
+ # self.assertFormal(FIFOModelEquivalenceSpec(fifo, rdomain="read", wdomain="write"),
+ # mode="bmc", depth=fifo.depth * 3 + 1)
+ self.assertFormal(FIFOContractSpec(fifo, rdomain="read", wdomain="write",
+ bound=fifo.depth * 4 + 1),
+ mode="hybrid", depth=fifo.depth * 4 + 1)
+
+ def test_async(self):
+ self.check_async_fifo(AsyncFIFO(width=8, depth=4))
+
+ def test_async_buffered(self):
+ self.check_async_fifo(AsyncFIFOBuffered(width=8, depth=3))
projects / nmigen.git / commitdiff