add big rx test and fix it

board/main_comms.h

@@ -61,13 +61,16 @@ bool add_magic = true;
 
 int comms_can_read(uint8_t *data, uint32_t max_len) {
   uint32_t pos = 0U;
+  bool added_magic = false;
 
   if (add_magic && (max_len >= sizeof(uint32_t))) {
     // Start of a transaction
     *((uint32_t *)(void *) &data[0]) = CAN_TRANSACTION_MAGIC;
     pos += sizeof(uint32_t);
     add_magic = false;
     can_read_buffer.ptr = 0U;
+    total_rx_size = 0U;
+    added_magic = true;
   }
 
   // Send tail of previous message if it is in buffer
@@ -79,11 +82,7 @@ int comms_can_read(uint8_t *data, uint32_t max_len) {
     can_read_buffer.ptr -= overflow_len;
   }
 
-  if (total_rx_size >= MAX_EP1_CHUNK_PER_BULK_TRANSFER) {
-    // We ran out of space for this transaction, prepare for the next one
-    total_rx_size = 0U;
-    add_magic = true;
-  } else {
+  if ((total_rx_size + pos) < MAX_EP1_CHUNK_PER_BULK_TRANSFER) {
     // Fill rest of buffer with new data
     CANPacket_t can_packet;
     while ((pos < max_len) && can_pop(&can_rx_q, &can_packet)) {
@@ -99,15 +98,19 @@ int comms_can_read(uint8_t *data, uint32_t max_len) {
         pos = max_len;
       }
     }
-    total_rx_size += pos;
   }
 
   if (pos != max_len) {
     // Final packet for this transaction, prepare for the next one
-    total_rx_size = 0U;
     add_magic = true;
   }
 
+  if (added_magic && (pos == sizeof(uint32_t))) {
+    // Magic alone doesn't make sense
+    pos = 0U;
+  }
+
+  total_rx_size += pos;
   return pos;
 }
 

tests/canfd/test_canfd.py

@@ -2,7 +2,6 @@
 import os
 import time
 import random
-import _thread
 from collections import defaultdict
 from panda import Panda
 from panda_jungle import PandaJungle  # pylint: disable=import-error
@@ -18,19 +17,9 @@
 
 _panda_serials = []
 
-def start_heartbeat_thread(p):
-  def heartbeat_thread(p):
-    while True:
-      try:
-        p.send_heartbeat()
-        time.sleep(.5)
-      except Exception:
-        break
-  _thread.start_new_thread(heartbeat_thread, (p,))
-
 def panda_init(serial, enable_canfd=False):
   p = Panda(serial=serial)
-  start_heartbeat_thread(p)
+  p.set_heartbeat_disabled()
   p.set_power_save(False)
   for bus in range(3):
     if enable_canfd:

tests/usbprotocol/libpandaprotocol_py.py

@@ -38,6 +38,7 @@
 bool can_push(can_ring *q, CANPacket_t *elem);
 int comms_can_read(uint8_t *data, uint32_t max_len);
 void comms_can_write(uint8_t *data, uint32_t len);
+uint32_t can_slots_empty(can_ring *q);
 """)
 
 class CANPacket:

tests/usbprotocol/test_comms.py

@@ -1,7 +1,7 @@
 import random
 import unittest
 
-from panda import LEN_TO_DLC, DLC_TO_LEN, pack_can_buffer
+from panda import LEN_TO_DLC, DLC_TO_LEN, pack_can_buffer, unpack_can_buffer
 from panda.tests.usbprotocol.libpandaprotocol_py import ffi, libpandaprotocol as lpp
 
 TX_QUEUES = (lpp.tx1_q, lpp.tx2_q, lpp.tx3_q, lpp.txgmlan_q)
@@ -19,12 +19,12 @@ def package_can_msg(msg):
 def unpackage_can_msg(pkt):
   dat_len = DLC_TO_LEN[pkt[0].data_len_code]
   dat = bytes(pkt[0].data[0:dat_len])
-  return pkt[0].addr, None, dat, pkt[0].bus
+  return pkt[0].addr, 0, dat, pkt[0].bus
 
 class PandaCommsTest(unittest.TestCase):
   def test_tx_queues(self):
     for bus in range(4):
-      message = (0x100, None, b"test", bus)
+      message = (0x100, 0, b"test", bus)
 
       can_pkt_tx = package_can_msg(message)
       can_pkt_rx = ffi.new('CANPacket_t *')
@@ -34,15 +34,14 @@ def test_tx_queues(self):
 
       assert unpackage_can_msg(can_pkt_rx) == message
 
-  def test_can_send(self):
+  def test_can_send_usb(self):
     for bus in range(3):
       for _ in range(100):
         msgs = []
         for _ in range(200):
           address = random.randint(1, 0x1FFFFFFF)
           data = bytes([random.getrandbits(8) for _ in range(DLC_TO_LEN[random.randrange(0, len(DLC_TO_LEN))])])
-          msgs.append((address, None, data, bus))
-
+          msgs.append((address, 0, data, bus))
         packed = pack_can_buffer(msgs)
 
         # Simulate USB bulk chunks
@@ -58,6 +57,41 @@ def test_can_send(self):
         while lpp.can_pop(TX_QUEUES[bus], pkt):
           queue_msgs.append(unpackage_can_msg(pkt))
 
-        assert len(queue_msgs) == len(msgs), f"{len(msgs)} {len(queue_msgs)}"
+        assert len(queue_msgs) == len(msgs), f"Expected: {len(msgs)} Queued: {len(queue_msgs)}"
+        assert queue_msgs == msgs
+
+  def test_can_receive_usb(self):
+    msgs = []
+    for _ in range(50000):
+      bus = random.randint(0, 3)
+      address = random.randint(1, 0x1FFFFFFF)
+      data = bytes([random.getrandbits(8) for _ in range(DLC_TO_LEN[random.randrange(0, len(DLC_TO_LEN))])])
+      msgs.append((address, 0, data, bus))
+    packets = list(map(package_can_msg, msgs))
+
+    rx_msgs = []
+    while len(packets) > 0:
+      # Push into queue
+      while lpp.can_slots_empty(lpp.rx_q) > 0 and len(packets) > 0:
+        lpp.can_push(lpp.rx_q, packets.pop(0))
+
+      # Simulate USB bulk IN chunks
+      CHUNK_SIZE = 0x40
+      MAX_TRANSFER_SIZE = 16384
+      dat = ffi.new(f"uint8_t[{CHUNK_SIZE}]")
+      while True:
+        buf = b""
+        while len(buf) < MAX_TRANSFER_SIZE:
+          max_size = min(CHUNK_SIZE, MAX_TRANSFER_SIZE - len(buf))
+          rx_len = lpp.comms_can_read(dat, max_size)
+          buf += bytes(dat[0:rx_len])
+          if rx_len < max_size:
+            break
+
+        if len(buf) == 0:
+          break
+        rx_msgs.extend(unpack_can_buffer(buf))
 
+    assert len(rx_msgs) == len(msgs), f"Expected: {len(msgs)} Received: {len(rx_msgs)}"
+    assert rx_msgs == msgs