From bee0c07685c63825fade18525028590527935fca Mon Sep 17 00:00:00 2001
From: Kia <kia@special-circumstanc.es>
Date: Sun, 14 Mar 2021 19:28:47 -0600
Subject: [PATCH] pipe_stage formal properties

---
 rtl_lib/pipe_stage.py | 240 +++++++++++++-----------------------------
 1 file changed, 76 insertions(+), 164 deletions(-)

diff --git a/rtl_lib/pipe_stage.py b/rtl_lib/pipe_stage.py
index f8c77b6..742a405 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()
-- 
GitLab