diff --git a/rtl_lib/pipe_stage.py b/rtl_lib/pipe_stage.py
index f8c77b60f1bacdab560cc0aba700dce91036dedf..742a405ef202b1cbc774a31fa6932a755b0e1239 100644
--- a/rtl_lib/pipe_stage.py
+++ b/rtl_lib/pipe_stage.py
@@ -6,10 +6,6 @@ from nmigen.cli import main
from nmigen.lib.fifo import *
-
-
-
-
class PipeStage(Elaboratable):
def __init__(self, *, width, registered_ready):
self.width = width
@@ -26,6 +22,39 @@ class PipeStage(Elaboratable):
def elaborate(self, platform):
m = Module()
+ # INTERFACE FORMAL PROPERTIES
+
+ # Stable Ready for the upstream interface cannot be guarantted if we have a register in
+ # the way (otherwise we'd need a combinatorial circuit which would defeat the point).
+ if(self.registered_ready == False):
+ with m.If((Past(self.upstream_ready_out) & (~Past(self.upstream_valid_in))) == 1):
+ m.d.comb += Assert(self.upstream_ready_out == 1)
+
+ # Assumption of Stable Ready on downstream interface is not needed for this proof
+# with m.If((Past(self.downstream_ready_in) & (~Past(self.downstream_valid_out))) == 1):
+# m.d.comb += Assume(self.downstream_ready_in == 1)
+
+
+ # Assumption of Stable data/valid on upstream interface is not needed for this proof:
+# with m.If((~Past(self.upstream_ready_out) & (Past(self.upstream_valid_in))) == 1):
+# m.d.comb += Assert(Stable(self.upstream_data_in) & self.upstream_valid_in)
+
+ with m.If(Past(ResetSignal()) == 0):
+ with m.If((~Past(self.downstream_ready_in) & (Past(self.downstream_valid_out))) == 1):
+ m.d.comb += Assert(Stable(self.downstream_data_out) & self.downstream_valid_out)
+
+
+ # Cover interface transactions
+ m.d.comb += Cover(self.upstream_ready_out & self.upstream_valid_in)
+ m.d.comb += Cover(self.downstream_ready_in & self.downstream_valid_out)
+
+ # Cover interface stalls
+
+ m.d.comb += Cover((~self.upstream_ready_out) & self.upstream_valid_in)
+ m.d.comb += Cover((~self.downstream_ready_in) & self.downstream_valid_out)
+
+
+
if (self.registered_ready == False):
m.d.comb += self.upstream_ready_out.eq(self.downstream_ready_in | (~self.downstream_valid_out))
@@ -59,191 +88,74 @@ class PipeStage(Elaboratable):
with m.If(self.downstream_ready_in == 1):
m.d.sync += skid_reg_valid.eq(0)
- #with m.If((Past(self.downstream_ready_in) & (~Past(self.downstream_valid_out))) == 1):
- # m.d.comb += Assume(self.downstream_ready_in == 1)
-
- #with m.If((Past(self.upstream_ready_out) & (~Past(self.upstream_valid_in))) == 1):
- # m.d.comb += Assert(self.upstream_ready_out == 1)
-
- number_captured = Signal(3)
+ # FORMAL SECTION
+ number_captured = Signal(3)
- upstream_transaction = Signal(1)
- downstream_transaction = Signal(1)
+ upstream_transaction = Signal(1)
+ downstream_transaction = Signal(1)
- m.d.comb += upstream_transaction.eq(self.upstream_ready_out & self.upstream_valid_in)
- m.d.comb += downstream_transaction.eq(self.downstream_ready_in & self.downstream_valid_out)
+ m.d.comb += upstream_transaction.eq(self.upstream_ready_out & self.upstream_valid_in)
+ m.d.comb += downstream_transaction.eq(self.downstream_ready_in & self.downstream_valid_out)
- m.d.sync += number_captured.eq(number_captured + upstream_transaction - downstream_transaction)
+ m.d.sync += number_captured.eq(number_captured + upstream_transaction - downstream_transaction)
- with m.If(pipe_reg_valid == 0):
- with m.If(skid_reg_valid == 0):
- m.d.comb += Assert(number_captured == 0)
+ with m.If(pipe_reg_valid == 0):
+ with m.If(skid_reg_valid == 0):
+ m.d.comb += Assert(number_captured == 0)
+ with m.Else():
+ m.d.comb += Assert(number_captured == 1)
with m.Else():
- m.d.comb += Assert(number_captured == 1)
- with m.Else():
- with m.If(skid_reg_valid == 0):
- m.d.comb += Assert(number_captured == 1)
- with m.Else():
- m.d.comb += Assert(number_captured == 2)
+ with m.If(skid_reg_valid == 0):
+ m.d.comb += Assert(number_captured == 1)
+ with m.Else():
+ m.d.comb += Assert(number_captured == 2)
- m.d.comb += Assert((number_captured == 0) | (number_captured == 1) | (number_captured == 2))
+ m.d.comb += Assert((number_captured == 0) | (number_captured == 1) | (number_captured == 2))
- regular_sample = Signal(self.width)
- skid_sample = Signal(self.width)
+ regular_sample = Signal(self.width)
+ skid_sample = Signal(self.width)
- with m.If(upstream_transaction == 1):
- with m.If(downstream_transaction == 1):
- # Upstream transaction, downstream transaction
- with m.If(number_captured == 2):
- m.d.comb += Assert(self.downstream_data_out == skid_sample)
- with m.Elif(number_captured == 1):
- m.d.comb += Assert(self.downstream_data_out == regular_sample)
- m.d.sync += regular_sample.eq(self.upstream_data_in)
+ with m.If(upstream_transaction == 1):
+ with m.If(downstream_transaction == 1):
+ # Upstream transaction, downstream transaction
+ with m.If(number_captured == 2):
+ m.d.comb += Assert(self.downstream_data_out == skid_sample)
+ with m.Elif(number_captured == 1):
+ m.d.comb += Assert(self.downstream_data_out == regular_sample)
+ m.d.sync += regular_sample.eq(self.upstream_data_in)
- with m.Else():
- # Upstream transaction, no downstream transaction
- with m.If(number_captured == 0):
- m.d.sync += regular_sample.eq(self.upstream_data_in)
- with m.Elif(number_captured == 1):
- m.d.sync += skid_sample.eq(regular_sample)
- m.d.sync += regular_sample.eq(self.upstream_data_in)
-
- with m.Else():
- with m.If(downstream_transaction == 1):
- # no upstream transaction, downstream transaction
- with m.If(number_captured == 2):
- m.d.comb += Assert(self.downstream_data_out == skid_sample)
- with m.Elif(number_captured == 1):
- m.d.comb += Assert(self.downstream_data_out == regular_sample)
+ with m.Else():
+ # Upstream transaction, no downstream transaction
+ with m.If(number_captured == 0):
+ m.d.sync += regular_sample.eq(self.upstream_data_in)
+ with m.Elif(number_captured == 1):
+ m.d.sync += skid_sample.eq(regular_sample)
+ m.d.sync += regular_sample.eq(self.upstream_data_in)
with m.Else():
- # no upstream transaction, no downstream transaction
- m.d.comb += Assert(1==1)
-
-
-
-
-
- with m.If(Past(ResetSignal()) | ResetSignal () == 1):
- m.d.comb += Assume(self.upstream_ready_out & self.upstream_valid_in == 0)
-
- with m.If(Past(ResetSignal()) == 0):
-
-
- with m.If((~Past(self.downstream_ready_in) & (Past(self.downstream_valid_out))) == 1):
- m.d.comb += Assert(Stable(self.downstream_data_out) & self.downstream_valid_out)
-
- return m
-
-
-
-
-class OldPipeStage(Elaboratable): # registered
- def __init__(self, width):
- self.width = width
-
- self.upstream_valid_in = Signal(1)
- self.upstream_ready_out = Signal(1)
- self.upstream_data_in = Signal(self.width)
-
- self.downstream_valid_out = Signal(1)
- self.downstream_ready_in = Signal(1)
- self.downstream_data_out = Signal(self.width)
-
- def elaborate(self, platform):
- m = Module()
-
- skid_buffer = Signal(self.width)
- skid_buffer_frozen = Signal(1)
-
- # we signal ready to upstream if and only if the skid buffer is empty
- m.d.comb += self.upstream_ready_out.eq(~skid_buffer_frozen)
-
-
- # should we capture a value in the skid buffer?
- # we should if upstream has data for us *and* downstream cannot accept
-
- # If we want to fill our skid buffer, we need:
- # 1. valid input data (upstream_valid_in == 1)
- # 2. a buffer that is empty (skid_buffer_frozen == 0)
- # 3. stalled downstream (downstream_ready_in == 0) & (downstream_valid_out == 1)
-
- with m.If(self.upstream_valid_in & (~skid_buffer_frozen) & (~self.downstream_ready_in) & (self.downstream_valid_out)):
- m.d.sync += skid_buffer_frozen.eq(1)
-
- # if downstream is accepting data again, we will flush our skid buffer
- with m.If(self.downstream_ready_in == 1):
- m.d.sync += skid_buffer_frozen.eq(0)
-
-
- # Stalled == (downstream_ready_in == 0 & downstream_valid_out == 1)
- # so not stalled = !(downstream_ready_in == 0 & downstream_valid_out == 1)
- # = downstream_ready_in == 1 | downstream_valid_out == 0
- # by de Morgan
-
- with m.If((self.downstream_ready_in) | (~self.downstream_valid_out)):
- m.d.sync += self.downstream_valid_out.eq(self.upstream_valid_in | skid_buffer_frozen)
-
- # data path
+ with m.If(downstream_transaction == 1):
+ # no upstream transaction, downstream transaction
+ with m.If(number_captured == 2):
+ m.d.comb += Assert(self.downstream_data_out == skid_sample)
+ with m.Elif(number_captured == 1):
+ m.d.comb += Assert(self.downstream_data_out == regular_sample)
- # always clock data into the skid buffer as long as the buffer isn't filled
+ with m.Else():
+ # no upstream transaction, no downstream transaction
+ m.d.comb += Assert(1==1)
- with m.If(skid_buffer_frozen == 0):
- m.d.sync += skid_buffer.eq(self.upstream_data_in)
-
- # not stalled condition
- with m.If((self.downstream_ready_in) | (~self.downstream_valid_out)):
- with m.If(skid_buffer_frozen == 1):
- m.d.sync += self.downstream_data_out.eq(skid_buffer)
- with m.Elif(self.upstream_valid_in == 1):
- m.d.sync += self.downstream_data_out.eq(self.upstream_data_in)
-
- return m
-class Trame(Elaboratable):
- def __init__(self, *, width, alpha, beta):
- self.width = width
- self.alpha = alpha
- self.beta = beta
-
- self.upstream_valid_in = Signal(1)
- self.upstream_ready_out = Signal(1)
- self.upstream_data_in = Signal(self.width)
-
- self.downstream_valid_out = Signal(1)
- self.downstream_ready_in = Signal(1)
- self.downstream_data_out = Signal(self.width)
-
- def elaborate(self, platform):
- m = Module()
-
- m.submodules += self.alpha
- m.submodules += self.beta
-
- m.d.comb += self.alpha.w_en.eq(self.upstream_valid_in)
- m.d.comb += self.alpha.w_data.eq(self.upstream_data_in)
- m.d.comb += self.alpha.r_en.eq(self.downstream_ready_in)
-
- m.d.comb += self.beta.upstream_valid_in.eq(self.upstream_valid_in)
- m.d.comb += self.beta.upstream_data_in.eq(self.upstream_data_in)
- m.d.comb += self.beta.downstream_ready_in.eq(self.downstream_ready_in)
-
- m.d.sync += Assert(self.beta.upstream_ready_out == self.alpha.w_rdy)
- m.d.sync += Assert(self.beta.downstream_valid_out == self.alpha.r_rdy)
- m.d.sync += Assert(self.beta.downstream_data_out == self.alpha.r_data)
-
return m
-
class DummyPlug(Elaboratable):
def elaborate(self, platform):
m = Module()