zephyr-cp: don't drop console input past 128 unread bytes

ports/zephyr-cp/common-hal/busio/UART.c

@@ -27,6 +27,14 @@
 #include <zephyr/logging/log.h>
 LOG_MODULE_REGISTER(busio_uart);
 
+// Turn RX back on after serial_cb() paused it for a full msgq.
+static void resume_rx(busio_uart_obj_t *self) {
+    if (self->rx_paused) {
+        self->rx_paused = false;
+        uart_irq_rx_enable(self->uart_device);
+    }
+}
+
 /*
  * Read characters from UART until line end is detected. Afterwards push the
  * data to the message queue.
@@ -42,20 +50,34 @@ static void serial_cb(const struct device *dev, void *user_data) {
         return;
     }
 
-    /* read until FIFO empty */
-    while (uart_fifo_read(dev, &c, 1) == 1) {
+    /* read until FIFO empty, or until msgq full if flow_control */
+    while (true) {
+        if (self->flow_control && k_msgq_num_free_get(&self->msgq) == 0) {
+            // Leave the remaining bytes in the driver instead of dropping
+            // them. USB CDC then stops accepting packets, so the host
+            // waits. read() calls resume_rx() after taking bytes out of
+            // the msgq.
+            self->rx_paused = true;
+            uart_irq_rx_disable(dev);
+            // read() may have freed msgq space just before rx_paused was
+            // set. Then read() saw rx_paused false and did not call
+            // resume_rx(), so check for space again.
+            if (k_msgq_num_free_get(&self->msgq) == 0) {
+                break;
+            }
+            resume_rx(self);
+        }
+        if (uart_fifo_read(dev, &c, 1) != 1) {
+            break;
+        }
         if (mp_interrupt_char == c) {
             common_hal_busio_uart_clear_rx_buffer(self);
             mp_sched_keyboard_interrupt();
             port_wake_main_task_from_isr();
-        } else if (!self->rx_paused) {
-            if (k_msgq_put(&self->msgq, &c, K_NO_WAIT) == 0) {
-                // Wake the main task so it can service the new RX data
-                // instead of sleeping out its full timeout.
-                port_wake_main_task_from_isr();
-            } else {
-                self->rx_paused = true;
-            }
+        } else if (k_msgq_put(&self->msgq, &c, K_NO_WAIT) == 0) {
+            // Wake the main task so it can service the new RX data
+            // instead of sleeping out its full timeout.
+            port_wake_main_task_from_isr();
         }
     }
 }
@@ -81,6 +103,7 @@ mp_obj_t common_hal_busio_uart_construct_from_device(busio_uart_obj_t *self, con
     self->timeout = K_FOREVER;
     self->write_timeout = K_FOREVER;
     self->rx_paused = false;
+    self->flow_control = false;
     uart_irq_rx_enable(uart_device);
 
     return MP_OBJ_FROM_PTR(self);
@@ -203,7 +226,7 @@ size_t common_hal_busio_uart_read(busio_uart_obj_t *self, uint8_t *data, size_t
         count++;
     }
     if (count > 0) {
-        self->rx_paused = false;
+        resume_rx(self);
     }
 
     return count;
@@ -253,6 +276,7 @@ uint32_t common_hal_busio_uart_rx_characters_available(busio_uart_obj_t *self) {
 
 void common_hal_busio_uart_clear_rx_buffer(busio_uart_obj_t *self) {
     k_msgq_purge(&self->msgq);
+    resume_rx(self);
 }
 
 bool common_hal_busio_uart_ready_to_tx(busio_uart_obj_t *self) {

ports/zephyr-cp/common-hal/busio/UART.h

@@ -27,6 +27,9 @@ typedef struct {
     const mcu_pin_obj_t *cts;
 
     bool rx_paused;     // set by irq if no space in rbuf
+    // Set when the sender waits for unread data (USB CDC), so RX can
+    // pause on a full msgq instead of dropping bytes.
+    bool flow_control;
 
     // True when the underlying Zephyr device was dynamically routed to the
     // pins above at construction time. Such objects own their receiver

ports/zephyr-cp/common-hal/usb_cdc/Serial.c

@@ -12,6 +12,7 @@
 
 mp_obj_t common_hal_usb_cdc_serial_construct_from_device(usb_cdc_serial_obj_t *self, const struct device *uart_device, uint16_t receiver_buffer_size, byte *receiver_buffer) {
     common_hal_busio_uart_construct_from_device(self, uart_device, receiver_buffer_size, receiver_buffer);
+    self->flow_control = true;
     self->base.type = &usb_cdc_serial_type;
     return MP_OBJ_FROM_PTR(self);
 }

ports/zephyr-cp/supervisor/usb.c

@@ -270,9 +270,9 @@ void usb_init(void) {
 
     bool console = usb_cdc_console_enabled();
     if (console) {
-        uint8_t *receiver_buffer = port_malloc(128, true);
+        uint8_t *receiver_buffer = port_malloc(1024, true);
         if (receiver_buffer != NULL) {
-            common_hal_usb_cdc_serial_construct_from_device(&usb_cdc_console_obj, DEVICE_DT_GET(DT_NODELABEL(cdc_acm_console)), 128, receiver_buffer);
+            common_hal_usb_cdc_serial_construct_from_device(&usb_cdc_console_obj, DEVICE_DT_GET(DT_NODELABEL(cdc_acm_console)), 1024, receiver_buffer);
         } else {
             console = false;
         }
@@ -281,9 +281,9 @@ void usb_init(void) {
 
     bool data = usb_cdc_data_enabled();
     if (data) {
-        uint8_t *receiver_buffer = port_malloc(128, true);
+        uint8_t *receiver_buffer = port_malloc(1024, true);
         if (receiver_buffer != NULL) {
-            common_hal_usb_cdc_serial_construct_from_device(&usb_cdc_data_obj, DEVICE_DT_GET(DT_NODELABEL(cdc_acm_data)), 128, receiver_buffer);
+            common_hal_usb_cdc_serial_construct_from_device(&usb_cdc_data_obj, DEVICE_DT_GET(DT_NODELABEL(cdc_acm_data)), 1024, receiver_buffer);
         } else {
             data = false;
         }

ports/zephyr-cp/tests/test_basics.py

@@ -124,6 +124,18 @@ def test_ctrl_c_interrupts_long_sleep(circuitpython):
     assert elapsed < 5, f"Ctrl+C took {elapsed:.1f}s to interrupt time.sleep(30)"
 
 
+@pytest.mark.circuitpy_drive({"code.py": 'print("spinning")\nwhile True:\n    pass\n'})
+@pytest.mark.duration(10)
+def test_ctrl_c_after_unread_input(circuitpython):
+    """Ctrl+C must still interrupt when the console queue is full of unread input."""
+    circuitpython.serial.wait_for("spinning")
+    circuitpython.serial.write("x" * 200)
+    time.sleep(0.5)
+    circuitpython.serial.write("\x03")
+    circuitpython.wait_until_done()
+    assert "KeyboardInterrupt" in circuitpython.serial.all_output
+
+
 RELOAD_CODE = """\
 print("first run")
 import time