drtio: add buffer space support to rt_packet_repeater

artiq/gateware/drtio/rt_packet_master.py

@@ -15,7 +15,7 @@ class RTPacketMaster(Module):
 
         # standard request interface
         #
-        # notwrite=1 address=0  buffer space request
+        # notwrite=1 address=0  buffer space request <destination>
         # notwrite=1 address=1  read request <channel, timestamp>
         #
         # optimized for write throughput
@@ -252,7 +252,7 @@ class RTPacketMaster(Module):
             )
         )
         tx_fsm.act("BUFFER_SPACE",
-            tx_dp.send("buffer_space_request"),
+            tx_dp.send("buffer_space_request", destination=sr_channel),
             If(tx_dp.packet_last,
                 sr_buf_re.eq(1),
                 NextState("IDLE")

artiq/gateware/drtio/rt_packet_repeater.py

@@ -1,14 +1,25 @@
 from migen import *
 from migen.genlib.fsm import *
+from migen.genlib.misc import WaitTimer
+
 
 from artiq.gateware.rtio import cri
+from artiq.gateware.drtio.cdc import CrossDomainNotification
 from artiq.gateware.drtio.rt_serializer import *
 
 
 class RTPacketRepeater(Module):
     def __init__(self, link_layer):
+        # CRI target interface in rtio domain
         self.cri = cri.Interface()
 
+        # in rtio_rx domain
+        self.err_unknown_packet_type = Signal()
+        self.err_packet_truncated = Signal()
+
+        # in rtio domain
+        self.buffer_space_timeout = Signal()
+
         # RX/TX datapath
         assert len(link_layer.tx_rt_data) == len(link_layer.rx_rt_data)
         assert len(link_layer.tx_rt_data) % 8 == 0
@@ -61,12 +72,30 @@ class RTPacketRepeater(Module):
                 extra_data_counter.eq(1)
             )
 
+        # Buffer space
+        buffer_space_destination = Signal(8)
+        self.sync.rtio += If(self.cri.cmd == cri.commands["get_buffer_space"],
+            buffer_space_destination.eq(self.cri.chan_sel[16:]))
+
+        rx_buffer_space_not = Signal()
+        rx_buffer_space = Signal(16)
+        buffer_space_not = Signal()
+        buffer_space_not_ack = Signal()
+        self.submodules += CrossDomainNotification("rtio_rx", "rtio",
+            rx_buffer_space_not, rx_buffer_space,
+            buffer_space_not, buffer_space_not_ack,
+            self.cri.o_buffer_space)
+
+        timeout_counter = WaitTimer(8191)
+        self.submodules += timeout_counter
+
         # TX FSM
         tx_fsm = ClockDomainsRenamer("rtio")(FSM(reset_state="IDLE"))
         self.submodules += tx_fsm
 
         tx_fsm.act("IDLE",
-            If(self.cri.cmd == cri.commands["write"], NextState("WRITE"))
+            If(self.cri.cmd == cri.commands["write"], NextState("WRITE")),
+            If(self.cri.cmd == cri.commands["get_buffer_space"], NextState("BUFFER_SPACE"))
         )
         tx_fsm.act("WRITE",
             tx_dp.send("write",
@@ -90,3 +119,52 @@ class RTPacketRepeater(Module):
                 NextState("IDLE")
             )
         )
+        tx_fsm.act("BUFFER_SPACE",
+            tx_dp.send("buffer_space_request", destination=buffer_space_destination),
+            If(tx_dp.packet_last,
+                buffer_space_not_ack.eq(1),
+                NextState("WAIT_BUFFER_SPACE")
+            )
+        )
+        tx_fsm.act("WAIT_BUFFER_SPACE",
+            timeout_counter.wait.eq(1),
+            If(timeout_counter.done,
+                self.buffer_space_timeout.eq(1),
+                NextState("IDLE")
+            ).Else(
+                If(buffer_space_not,
+                    self.cri.o_buffer_space_valid.eq(1),
+                    NextState("IDLE")
+                ),
+            )
+        )
+
+        # RX FSM
+        rx_fsm = ClockDomainsRenamer("rtio_rx")(FSM(reset_state="INPUT"))
+        self.submodules += rx_fsm
+
+        ongoing_packet_next = Signal()
+        ongoing_packet = Signal()
+        self.sync.rtio_rx += ongoing_packet.eq(ongoing_packet_next)
+
+        rx_fsm.act("INPUT",
+            If(rx_dp.frame_r,
+                rx_dp.packet_buffer_load.eq(1),
+                If(rx_dp.packet_last,
+                    Case(rx_dp.packet_type, {
+                        rx_plm.types["buffer_space_reply"]: NextState("BUFFER_SPACE"),
+                        "default": self.err_unknown_packet_type.eq(1)
+                    })
+                ).Else(
+                    ongoing_packet_next.eq(1)
+                )
+            ),
+            If(~rx_dp.frame_r & ongoing_packet,
+                self.err_packet_truncated.eq(1)
+            )
+        )
+        rx_fsm.act("BUFFER_SPACE",
+            rx_buffer_space_not.eq(1),
+            rx_buffer_space.eq(rx_dp.packet_as["buffer_space_reply"].space),
+            NextState("INPUT")
+        )

artiq/gateware/drtio/rt_serializer.py

@@ -53,7 +53,7 @@ def get_m2s_layouts(alignment):
                           ("address", 16),
                           ("extra_data_cnt", 8),
                           ("short_data", short_data_len))
-    plm.add_type("buffer_space_request")
+    plm.add_type("buffer_space_request", ("destination", 8))
 
     plm.add_type("read_request", ("channel", 16), ("timeout", 64))
 

artiq/gateware/test/drtio/test_rt_packet_repeater.py

@@ -60,3 +60,39 @@ class TestRepeater(unittest.TestCase):
 
             run_simulation(dut, [send(), pr.receive(receive)])
             self.assertEqual(test_writes, received)
+
+    def test_buffer_space(self):
+        for nwords in range(1, 8):
+            pt, pr, dut = create_dut(nwords)
+
+            def send_requests():
+                for i in range(10):
+                    yield dut.cri.chan_sel.eq(i << 16)
+                    yield dut.cri.cmd.eq(cri.commands["get_buffer_space"])
+                    yield
+                    yield dut.cri.cmd.eq(cri.commands["nop"])
+                    yield
+                    while not (yield dut.cri.o_buffer_space_valid):
+                        yield
+                    buffer_space = yield dut.cri.o_buffer_space
+                    self.assertEqual(buffer_space, 2*i)
+
+            current_request = None
+
+            @passive
+            def send_replies():
+                nonlocal current_request
+                while True:
+                    while current_request is None:
+                        yield
+                    yield from pt.send("buffer_space_reply", space=2*current_request)
+                    current_request = None
+
+            def receive(packet_type, field_dict, trailer):
+                nonlocal current_request
+                self.assertEqual(packet_type, "buffer_space_request")
+                self.assertEqual(trailer, [])
+                self.assertEqual(current_request, None)
+                current_request = field_dict["destination"]
+
+            run_simulation(dut, [send_requests(), send_replies(), pr.receive(receive)])