Pad Not TCP with zeros

In testing on real hardware (Fomu), I was missing a lot of "tail end"
packets... like, a *lot* of them. This prevented clients from getting
the full body, hanging up, etc etc.

This seems to be related to the 64B packet size of the USB ACM CDC TLA
WTF protocol we're using under the hood. I haven't checked the gateware,
but it's *acting like* it doesn't send the packet until there's >64B in
the buffer. (Or maybe 64 exactly, and I was doing my math wrong before.)

Luckily, we haven't been using "Stream ID zero", so we can redefine Not
TCP such that every packet begins with a nonzero byte (stream ID), and
zero bytes between packets are ignored. Then, pad on the device side,
and *wham*, working.

(Mostly. The LED endpoint doesn't always want to finish its job; needs
more investigation.)

Author: cceckman

not_tcp/host.py

@@ -79,7 +79,7 @@ def from_header(cls, header: Header, body: bytes) -> "Packet":
         )
 
     def __len__(self):
-        return len(Header) + len(self.body)
+        return Header.length() + len(self.body)
 
     def header(self) -> Header:
         assert self.stream_id >= 0
@@ -92,6 +92,11 @@ def to_bytes(self) -> bytes:
 
     @classmethod
     def from_bytes(cls, buf: bytes) -> (Optional["Packet"], bytes):
+        # Trim null bytes that prefix a packet.
+        # This is (retroactively) why we start with stream ID 1!
+        while len(buf) > 0 and buf[0] == 0:
+            buf = buf[1:]
+
         if len(buf) < Header.length():
             return None, buf
         header = Header.from_bytes(buf[:Header.length()])
@@ -171,17 +176,30 @@ async def run_outbound(self, number: int, writer: StreamWriter):
             consumed = (buffer_len - len(buffer))
             total_bytes += consumed
             if p is None:
-                continue
-            buffer = rem
-            if packet_count == 0:
-                assert p.start
-            packet_count += 1
-            if not p.to_host:
-                # Ignore the packet
-                continue
-            writer.write(p.body)
-            await writer.drain()
-            if p.end:
-                break
+                if consumed > 0:
+                    olog.debug(
+                        f"consumed {consumed} padding zeros")
+
+                # No packet to consume. Get more data.
+                rcvd = self.recv()  # Has its own timeout, but isn't async. So:
+                await asyncio.sleep(0)
+                buffer += rcvd
+            else:
+                olog.debug(f"consumed {consumed} bytes")
+
+                if packet_count == 0:
+                    assert p.start
+                packet_count += 1
+                if not p.to_host:
+                    # Ignore the packet
+                    continue
+                writer.write(p.body)
+                await writer.drain()
+                if p.end:
+                    break
+        olog.info(
+            f"device closed outbound connection for client {number}")
+
+        # TODO: Handle multiple streams.
         writer.close()
         await writer.wait_closed()

not_tcp/not_tcp.py

@@ -15,6 +15,8 @@
 import session
 from stream_utils import LimitForwarder
 
+BUFFER_SIZE = 256
+
 
 class Flags(Struct):
     """
@@ -119,7 +121,7 @@ def elaborate(self, platform):
         m = Module()
         connected = self.connected
         input_buffer = m.submodules.input_buffer = SyncFIFOBuffered(
-            width=8, depth=256)
+            width=8, depth=BUFFER_SIZE)
         m.d.comb += [
             self.stop.data.payload.eq(input_buffer.r_stream.payload),
             self.stop.data.valid.eq(input_buffer.r_stream.valid),
@@ -236,9 +238,8 @@ def __init__(self, stream_id):
     def elaborate(self, platform):
         m = Module()
 
-        # Each of these is big enough to buffer one full packet.
         output_buffer = m.submodules.output_buffer = SyncFIFOBuffered(
-            width=8, depth=256)
+            width=8, depth=BUFFER_SIZE)
 
         m.d.comb += [
             output_buffer.w_stream.payload.eq(self.stop.data.payload),
@@ -254,23 +255,33 @@ def elaborate(self, platform):
             output_limiter.start.eq(0),
             self.bus.valid.eq(0)
         ]
-        connect(m, output_buffer.r_stream, output_limiter.inbound)
-
+        m.d.comb += [
+            self.bus.payload.eq(output_limiter.outbound.payload),
+            self.bus.valid.eq(output_limiter.outbound.valid),
+            output_limiter.outbound.ready.eq(self.bus.ready),
+        ]
         flags_layout = UnionLayout({"bytes": unsigned(8), "flags": Flags})
 
         # Flags for outbound packet:
         send_flags = Signal(flags_layout)
         m.d.sync += send_flags.flags.to_host.eq(1)
         send_len = Signal(8)
+        # Pad up to 64 bytes of zeros, to ensure packet delivery.
+        pad_len = Signal(8)
+
+        # Invariants:
+        # - End is set iff ~active and the buffer is empty.
+        # - We enter disconnected iff End is clear,
+        #   i.e. End has been sent.
 
         # Cases in which we want to send a packet:
         with m.FSM(name="write"):
             with m.State("disconnected"):
+                m.d.comb += Assert(~send_flags.flags.end)
                 m.next = "disconnected"
                 with m.If(self.stop.active):
                     # Immediately send a "start" packet.
                     m.d.sync += send_flags.flags.start.eq(1)
-                    m.d.sync += send_flags.flags.end.eq(0)
                     m.d.sync += self.connected.eq(1)
                     m.next = "write-stream"
 
@@ -287,17 +298,27 @@ def elaborate(self, platform):
                     # Lock in the level as the length of this packet.
                     # We may send a short (zero-length) packet
                     # to start or end the connection.
-                    m.d.sync += send_len.eq(output_buffer.r_level)
+                    m.d.sync += send_len.eq(output_buffer.level)
                     # We send an explicit empty END packet.
                     m.d.sync += send_flags.flags.end.eq(
                         ~self.stop.active &
-                        (output_buffer.r_level == Const(0)))
+                        (output_buffer.level == Const(0)))
                     with m.If(self.bus.ready):
                         m.next = "write-len"
             with m.State("write-len"):
                 m.next = "write-len"
                 m.d.comb += self.bus.payload.eq(send_len)
                 m.d.comb += self.bus.valid.eq(1)
+
+                # Precompute padding on a cycle where we don't otherwise
+                # have much to do.
+                # In theory we only need to pad to 64...
+                # but that still doesn't get us the stop byte...
+                # so, double-padding?
+                m.d.sync += [
+                    pad_len.eq(128 - ((3 + send_len) % 64))
+                ]
+
                 with m.If(self.bus.ready):
                     m.next = "write-flags"
             with m.State("write-flags"):
@@ -314,22 +335,35 @@ def elaborate(self, platform):
                     m.next = "write-body"
             with m.State("write-body"):
                 m.next = "write-body"
+                connect(m, output_buffer.r_stream, output_limiter.inbound)
+
+                with m.If(output_limiter.done):
+                    m.next = "zero-pad"
+                    m.d.comb += [
+                        output_limiter.count.eq(pad_len),
+                        output_limiter.start.eq(1),
+                        output_limiter.inbound.payload.eq(0),
+                        output_limiter.inbound.valid.eq(1),
+                    ]
+
+            with m.State("zero-pad"):
+                m.next = "zero-pad"
                 m.d.comb += [
-                    self.bus.payload.eq(output_limiter.outbound.payload),
-                    self.bus.valid.eq(output_limiter.outbound.valid),
-                    output_limiter.outbound.ready.eq(self.bus.ready),
+                    output_limiter.inbound.payload.eq(0),
+                    output_limiter.inbound.valid.eq(1),
                 ]
 
                 with m.If(output_limiter.done):
-                    m.d.sync += [
-                        send_flags.flags.start.eq(0),
-                        send_flags.flags.end.eq(0),
-                    ]
                     with m.If(send_flags.flags.end):
                         m.d.sync += self.connected.eq(0)
                         m.next = "disconnected"
                     with m.Else():
                         m.next = "write-stream"
+                    # In either branch, we've sent a start packet.
+                    m.d.sync += [
+                        send_flags.flags.start.eq(0),
+                        send_flags.flags.end.eq(0),
+                    ]
 
         return m
 

not_tcp/not_tcp_test.py

@@ -78,8 +78,9 @@ async def driver(ctx):
     while len(rcvd) > 0:
         # All data should be packetized.
         (p, remainder) = Packet.from_bytes(rcvd)
-        assert p is not None, f"remaining data: {rcvd}"
-        packets += [p]
+        assert (p is not None) or (len(remainder) == 0)
+        if p is not None:
+            packets += [p]
         rcvd = remainder
     bodies = bytes()
     for i in range(len(packets)):

stream_fixtures.py

@@ -81,17 +81,21 @@ def collect_queue(self, q: queue.Queue[bytes], batch_size: int = 100,
         stream = self._stream
 
         async def collector(ctx):
-            ctx.set(stream.ready, 1)
+            ready = self.is_ready()
+            ctx.set(stream.ready, ready)
             countup = 0
             batch = bytes()
 
             async for clk_edge, rst_value, valid, payload in ctx.tick().sample(
                     stream.valid, stream.payload):
                 if rst_value or (not clk_edge):
                     continue
-                if valid == 1:
+                if ready == 1 and valid == 1:
                     # We just transferred a payload byte.
                     batch += bytes([payload])
+                    ready = self.is_ready()
+                else:
+                    ready = ready | self.is_ready()
                 countup += 1
 
                 batch_exceeded = len(batch) >= batch_size