sim: refactor dispatcher to crossbar

sim_pipeline.py

@@ -128,29 +128,31 @@ class Pixel_Decoder(Module):
 
         # TODO: support mono16 for now?
 
-class Streams_Dispatcher(Module):
+class Streams_Crossbar(Module):
-    def __init__(self, downconns):
+    def __init__(self, downconns, streams_buffer):
         n_downconn = len(downconns)
         self.submodules.mux = mux = stream.Multiplexer(word_layout_dchar, n_downconn)
 
         for i, c in enumerate(downconns):
-            # if i == 0:
             self.comb += [
-                # no backpressure
-                c.source.ack.eq(1),
                 c.source.connect(getattr(mux, "sink"+str(i)))
             ]
 
-        self.source = stream.Endpoint(word_layout_dchar)
-
         self.submodules.fsm = fsm = FSM(reset_state="WAIT_HEADER")
 
-        # TODO: add different downstream
+        self.stream_id = Signal(char_width)
-        # stream_id = Signal()
+        case = dict((i, mux.source.connect(b.sink)) for i, b in enumerate(streams_buffer))
-        # case = dict((i, mux.source.connect(b.sink)) for i, b in enumerate(buffers))
         fsm.act(
             "WAIT_HEADER",
-            mux.source.connect(self.source),
+            NextValue(self.stream_id, mux.source.dchar),
+            If(mux.source.stb,
+                NextState("STREAM"),
+            ),
+        )
+
+        fsm.act(
+            "STREAM",
+            Case(self.stream_id, case),
             If(mux.source.eop,
                 NextState("SWITCH_CONN"),
             ),

sim_stream.py

@@ -10,23 +10,31 @@ class CXP_Links(Module):
         # NOTE: although there are double buffer in each connect, the reading must be faster than writing to avoid data loss
 
         self.downconns = []
+        self.stream_buffers = []
         for i in range(2):
             downconn = Pipeline()
             setattr(self.submodules, "cxp_conn"+str(i), downconn)
             self.downconns.append(downconn)
 
-        self.submodules.dispatcher = dispatcher = Streams_Dispatcher(self.downconns)
+            double_buffer = Double_Stream_Buffer()
+            setattr(self.submodules, "stream_buffer"+str(i), double_buffer)
+            self.stream_buffers.append(double_buffer)
+
+            # no backpressure for sim purposes
+            self.sync += double_buffer.source.ack.eq(1)
+            
+
+        self.submodules.crossbar = crossbar = Streams_Crossbar(self.downconns, self.stream_buffers)
         
         # TODO: add extractor
-        # self.submodules.double_buffer = double_buffer = Double_Stream_Buffer()
 
-        pipeline = [dispatcher]
+        # pipeline = [dispatcher, double_buffer]
-        for s, d in zip(pipeline, pipeline[1:]):
+        # for s, d in zip(pipeline, pipeline[1:]):
-            self.comb += s.source.connect(d.sink)
+        #     self.comb += s.source.connect(d.sink)
-        self.source = pipeline[-1].source
+        # self.source = pipeline[-1].source
         
-        # no backpressure
+        # # no backpressure
-        self.sync += self.source.ack.eq(1)
+        # self.sync += self.source.ack.eq(1)
 
 
 class Pipeline(Module):
@@ -50,11 +58,10 @@ class Pipeline(Module):
 dut = CXP_Links()
 
 def check_case(packet=[]):
-        
-
-            
     print("=================TEST========================")
+
     downconns = dut.downconns
+    stream_buffers = dut.stream_buffers
     ch = 0
     
     for i, p in enumerate(packet):
@@ -85,11 +92,11 @@ def check_case(packet=[]):
 
         # check cycle result
         yield
-        source = dut.dispatcher.mux.source
+        source = dut.stream_buffers[0].source
         print(
             f"\nCYCLE#{i} : source char = {yield source.data:#X} k = {yield source.k:#X} stb = {yield source.stb} ack = {yield source.ack} eop = {yield source.eop}"
             # f" source dchar = {yield source.dchar:#X} dchar_k = {yield source.dchar_k:#X}"
-            f"\nCYCLE#{i} : read mask = {yield dut.dispatcher.mux.sel}"
+            f"\nCYCLE#{i} : read mask = {yield dut.crossbar .mux.sel}"
             # f"\nCYCLE#{i} : stream id = {yield decoder.stream_id:#X} pak_tag = {yield decoder.pak_tag:#X}"
             # f" stream_pak_size = {yield decoder.stream_pak_size:#X}"
         )
@@ -114,7 +121,7 @@ def check_case(packet=[]):
         print(
             f"\nCYCLE#{i} : source char = {yield source.data:#X} k = {yield source.k:#X} stb = {yield source.stb} ack = {yield source.ack} eop = {yield source.eop}"
             # f" source dchar = {yield source.dchar:#X} dchar_k = {yield source.dchar_k:#X}"
-            f"\nCYCLE#{i} : read mask = {yield dut.dispatcher.mux.sel}"
+            f"\nCYCLE#{i} : read mask = {yield dut.crossbar .mux.sel}"
             # f"\nCYCLE#{i} : stream id = {yield decoder.stream_id:#X} pak_tag = {yield decoder.pak_tag:#X}"
             # f" stream_pak_size = {yield decoder.stream_pak_size:#X}"
         )
@@ -122,7 +129,7 @@ def check_case(packet=[]):
 
 
 def testbench():
-    stream_id = 0x69
+    # stream_id = 0x01
     streams = [
         [
             {"data": 0x11111111, "k": Replicate(0, 4)},
@@ -142,7 +149,7 @@ def testbench():
     for i, s in enumerate(streams):
         s[-1]["eop"] = 0
         packet += [
-            {"data": Replicate(C(stream_id, char_width), 4), "k": Replicate(0, 4)},
+            {"data": Replicate(C(i % 2, char_width), 4), "k": Replicate(0, 4)},
             {"data": Replicate(C(i, char_width), 4), "k": Replicate(0, 4)},
             {
                 "data": Replicate(C(len(s) >> 8 & 0xFF, char_width), 4),