From 11ccf0e591465a69c74b488d7c3962295c51b669 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 24 Sep 2025 08:58:42 -0500 Subject: [PATCH] [usb_host] Prevent USB data corruption from missed events (#10859) --- esphome/components/usb_host/usb_host.h | 50 +++++++- .../components/usb_host/usb_host_client.cpp | 117 ++++++++++++++++-- esphome/components/usb_uart/ch34x.cpp | 6 +- esphome/components/usb_uart/cp210x.cpp | 4 +- esphome/components/usb_uart/usb_uart.cpp | 106 +++++++++++----- esphome/components/usb_uart/usb_uart.h | 34 ++++- 6 files changed, 266 insertions(+), 51 deletions(-) diff --git a/esphome/components/usb_host/usb_host.h b/esphome/components/usb_host/usb_host.h index c5466eb1f0..d2ff4da068 100644 --- a/esphome/components/usb_host/usb_host.h +++ b/esphome/components/usb_host/usb_host.h @@ -5,7 +5,10 @@ #include "esphome/core/component.h" #include #include "usb/usb_host.h" - +#include +#include +#include "esphome/core/lock_free_queue.h" +#include "esphome/core/event_pool.h" #include namespace esphome { @@ -13,6 +16,10 @@ namespace usb_host { static const char *const TAG = "usb_host"; +// Forward declarations +struct TransferRequest; +class USBClient; + // constants for setup packet type static const uint8_t USB_RECIP_DEVICE = 0; static const uint8_t USB_RECIP_INTERFACE = 1; @@ -25,7 +32,10 @@ static const uint8_t USB_DIR_IN = 1 << 7; static const uint8_t USB_DIR_OUT = 0; static const size_t SETUP_PACKET_SIZE = 8; -static const size_t MAX_REQUESTS = 16; // maximum number of outstanding requests possible. +static const size_t MAX_REQUESTS = 16; // maximum number of outstanding requests possible. +static constexpr size_t USB_EVENT_QUEUE_SIZE = 32; // Size of event queue between USB task and main loop +static constexpr size_t USB_TASK_STACK_SIZE = 4096; // Stack size for USB task (same as ESP-IDF USB examples) +static constexpr UBaseType_t USB_TASK_PRIORITY = 5; // Higher priority than main loop (tskIDLE_PRIORITY + 5) // used to report a transfer status struct TransferStatus { @@ -49,6 +59,31 @@ struct TransferRequest { USBClient *client; }; +enum EventType : uint8_t { + EVENT_DEVICE_NEW, + EVENT_DEVICE_GONE, + EVENT_TRANSFER_COMPLETE, + EVENT_CONTROL_COMPLETE, +}; + +struct UsbEvent { + EventType type; + union { + struct { + uint8_t address; + } device_new; + struct { + usb_device_handle_t handle; + } device_gone; + struct { + TransferRequest *trq; + } transfer; + } data; + + // Required for EventPool - no cleanup needed for POD types + void release() {} +}; + // callback function type. enum ClientState { @@ -84,6 +119,11 @@ class USBClient : public Component { bool control_transfer(uint8_t type, uint8_t request, uint16_t value, uint16_t index, const transfer_cb_t &callback, const std::vector &data = {}); + // Lock-free event queue and pool for USB task to main loop communication + // Must be public for access from static callbacks + LockFreeQueue event_queue; + EventPool event_pool; + protected: bool register_(); TransferRequest *get_trq_(); @@ -91,6 +131,12 @@ class USBClient : public Component { virtual void on_connected() {} virtual void on_disconnected() { this->init_pool(); } + // USB task management + static void usb_task_fn(void *arg); + void usb_task_loop(); + + TaskHandle_t usb_task_handle_{nullptr}; + usb_host_client_handle_t handle_{}; usb_device_handle_t device_handle_{}; int device_addr_{-1}; diff --git a/esphome/components/usb_host/usb_host_client.cpp b/esphome/components/usb_host/usb_host_client.cpp index 4c0c12fa18..5c9d56c7f9 100644 --- a/esphome/components/usb_host/usb_host_client.cpp +++ b/esphome/components/usb_host/usb_host_client.cpp @@ -139,24 +139,40 @@ static std::string get_descriptor_string(const usb_str_desc_t *desc) { return {buffer}; } +// CALLBACK CONTEXT: USB task (called from usb_host_client_handle_events in USB task) static void client_event_cb(const usb_host_client_event_msg_t *event_msg, void *ptr) { auto *client = static_cast(ptr); + + // Allocate event from pool + UsbEvent *event = client->event_pool.allocate(); + if (event == nullptr) { + // No events available - increment counter for periodic logging + client->event_queue.increment_dropped_count(); + return; + } + + // Queue events to be processed in main loop switch (event_msg->event) { case USB_HOST_CLIENT_EVENT_NEW_DEV: { - auto addr = event_msg->new_dev.address; ESP_LOGD(TAG, "New device %d", event_msg->new_dev.address); - client->on_opened(addr); + event->type = EVENT_DEVICE_NEW; + event->data.device_new.address = event_msg->new_dev.address; break; } case USB_HOST_CLIENT_EVENT_DEV_GONE: { - client->on_removed(event_msg->dev_gone.dev_hdl); - ESP_LOGD(TAG, "Device gone %d", event_msg->new_dev.address); + ESP_LOGD(TAG, "Device gone"); + event->type = EVENT_DEVICE_GONE; + event->data.device_gone.handle = event_msg->dev_gone.dev_hdl; break; } default: ESP_LOGD(TAG, "Unknown event %d", event_msg->event); - break; + client->event_pool.release(event); + return; } + + // Push to lock-free queue (always succeeds since pool size == queue size) + client->event_queue.push(event); } void USBClient::setup() { usb_host_client_config_t config{.is_synchronous = false, @@ -173,9 +189,59 @@ void USBClient::setup() { usb_host_transfer_alloc(64, 0, &trq->transfer); trq->client = this; } + + // Create and start USB task + xTaskCreate(usb_task_fn, "usb_task", + USB_TASK_STACK_SIZE, // Stack size + this, // Task parameter + USB_TASK_PRIORITY, // Priority (higher than main loop) + &this->usb_task_handle_); + + if (this->usb_task_handle_ == nullptr) { + ESP_LOGE(TAG, "Failed to create USB task"); + this->mark_failed(); + } +} + +void USBClient::usb_task_fn(void *arg) { + auto *client = static_cast(arg); + client->usb_task_loop(); +} + +void USBClient::usb_task_loop() { + while (true) { + usb_host_client_handle_events(this->handle_, portMAX_DELAY); + } } void USBClient::loop() { + // Process any events from the USB task + UsbEvent *event; + while ((event = this->event_queue.pop()) != nullptr) { + switch (event->type) { + case EVENT_DEVICE_NEW: + this->on_opened(event->data.device_new.address); + break; + case EVENT_DEVICE_GONE: + this->on_removed(event->data.device_gone.handle); + break; + case EVENT_TRANSFER_COMPLETE: + case EVENT_CONTROL_COMPLETE: { + auto *trq = event->data.transfer.trq; + this->release_trq(trq); + break; + } + } + // Return event to pool for reuse + this->event_pool.release(event); + } + + // Log dropped events periodically + uint16_t dropped = this->event_queue.get_and_reset_dropped_count(); + if (dropped > 0) { + ESP_LOGW(TAG, "Dropped %u USB events due to queue overflow", dropped); + } + switch (this->state_) { case USB_CLIENT_OPEN: { int err; @@ -228,7 +294,6 @@ void USBClient::loop() { } default: - usb_host_client_handle_events(this->handle_, 0); break; } } @@ -245,6 +310,26 @@ void USBClient::on_removed(usb_device_handle_t handle) { } } +// Helper to queue transfer cleanup to main loop +static void queue_transfer_cleanup(TransferRequest *trq, EventType type) { + auto *client = trq->client; + + // Allocate event from pool + UsbEvent *event = client->event_pool.allocate(); + if (event == nullptr) { + // No events available - increment counter for periodic logging + client->event_queue.increment_dropped_count(); + return; + } + + event->type = type; + event->data.transfer.trq = trq; + + // Push to lock-free queue (always succeeds since pool size == queue size) + client->event_queue.push(event); +} + +// CALLBACK CONTEXT: USB task (called from usb_host_client_handle_events in USB task) static void control_callback(const usb_transfer_t *xfer) { auto *trq = static_cast(xfer->context); trq->status.error_code = xfer->status; @@ -252,9 +337,14 @@ static void control_callback(const usb_transfer_t *xfer) { trq->status.endpoint = xfer->bEndpointAddress; trq->status.data = xfer->data_buffer; trq->status.data_len = xfer->actual_num_bytes; - if (trq->callback != nullptr) + + // Execute callback in USB task context + if (trq->callback != nullptr) { trq->callback(trq->status); - trq->client->release_trq(trq); + } + + // Queue cleanup to main loop + queue_transfer_cleanup(trq, EVENT_CONTROL_COMPLETE); } TransferRequest *USBClient::get_trq_() { @@ -315,6 +405,7 @@ bool USBClient::control_transfer(uint8_t type, uint8_t request, uint16_t value, return true; } +// CALLBACK CONTEXT: USB task (called from usb_host_client_handle_events in USB task) static void transfer_callback(usb_transfer_t *xfer) { auto *trq = static_cast(xfer->context); trq->status.error_code = xfer->status; @@ -322,9 +413,15 @@ static void transfer_callback(usb_transfer_t *xfer) { trq->status.endpoint = xfer->bEndpointAddress; trq->status.data = xfer->data_buffer; trq->status.data_len = xfer->actual_num_bytes; - if (trq->callback != nullptr) + + // Always execute callback in USB task context + // Callbacks should be fast and non-blocking (e.g., copy data to queue) + if (trq->callback != nullptr) { trq->callback(trq->status); - trq->client->release_trq(trq); + } + + // Queue cleanup to main loop + queue_transfer_cleanup(trq, EVENT_TRANSFER_COMPLETE); } /** * Performs a transfer input operation. diff --git a/esphome/components/usb_uart/ch34x.cpp b/esphome/components/usb_uart/ch34x.cpp index 74e7933824..601bfe7366 100644 --- a/esphome/components/usb_uart/ch34x.cpp +++ b/esphome/components/usb_uart/ch34x.cpp @@ -16,12 +16,12 @@ using namespace bytebuffer; void USBUartTypeCH34X::enable_channels() { // enable the channels for (auto channel : this->channels_) { - if (!channel->initialised_) + if (!channel->initialised_.load()) continue; usb_host::transfer_cb_t callback = [=](const usb_host::TransferStatus &status) { if (!status.success) { ESP_LOGE(TAG, "Control transfer failed, status=%s", esp_err_to_name(status.error_code)); - channel->initialised_ = false; + channel->initialised_.store(false); } }; @@ -48,7 +48,7 @@ void USBUartTypeCH34X::enable_channels() { auto factor = static_cast(clk / baud_rate); if (factor == 0 || factor == 0xFF) { ESP_LOGE(TAG, "Invalid baud rate %" PRIu32, baud_rate); - channel->initialised_ = false; + channel->initialised_.store(false); continue; } if ((clk / factor - baud_rate) > (baud_rate - clk / (factor + 1))) diff --git a/esphome/components/usb_uart/cp210x.cpp b/esphome/components/usb_uart/cp210x.cpp index f7d60c307a..35834c7529 100644 --- a/esphome/components/usb_uart/cp210x.cpp +++ b/esphome/components/usb_uart/cp210x.cpp @@ -100,12 +100,12 @@ std::vector USBUartTypeCP210X::parse_descriptors(usb_device_handle_t dev void USBUartTypeCP210X::enable_channels() { // enable the channels for (auto channel : this->channels_) { - if (!channel->initialised_) + if (!channel->initialised_.load()) continue; usb_host::transfer_cb_t callback = [=](const usb_host::TransferStatus &status) { if (!status.success) { ESP_LOGE(TAG, "Control transfer failed, status=%s", esp_err_to_name(status.error_code)); - channel->initialised_ = false; + channel->initialised_.store(false); } }; this->control_transfer(USB_VENDOR_IFC | usb_host::USB_DIR_OUT, IFC_ENABLE, 1, channel->index_, callback); diff --git a/esphome/components/usb_uart/usb_uart.cpp b/esphome/components/usb_uart/usb_uart.cpp index bf1c9086f1..8603e28d62 100644 --- a/esphome/components/usb_uart/usb_uart.cpp +++ b/esphome/components/usb_uart/usb_uart.cpp @@ -130,7 +130,7 @@ size_t RingBuffer::pop(uint8_t *data, size_t len) { return len; } void USBUartChannel::write_array(const uint8_t *data, size_t len) { - if (!this->initialised_) { + if (!this->initialised_.load()) { ESP_LOGV(TAG, "Channel not initialised - write ignored"); return; } @@ -152,7 +152,7 @@ bool USBUartChannel::peek_byte(uint8_t *data) { return true; } bool USBUartChannel::read_array(uint8_t *data, size_t len) { - if (!this->initialised_) { + if (!this->initialised_.load()) { ESP_LOGV(TAG, "Channel not initialised - read ignored"); return false; } @@ -170,7 +170,34 @@ bool USBUartChannel::read_array(uint8_t *data, size_t len) { return status; } void USBUartComponent::setup() { USBClient::setup(); } -void USBUartComponent::loop() { USBClient::loop(); } +void USBUartComponent::loop() { + USBClient::loop(); + + // Process USB data from the lock-free queue + UsbDataChunk *chunk; + while ((chunk = this->usb_data_queue_.pop()) != nullptr) { + auto *channel = chunk->channel; + +#ifdef USE_UART_DEBUGGER + if (channel->debug_) { + uart::UARTDebug::log_hex(uart::UART_DIRECTION_RX, std::vector(chunk->data, chunk->data + chunk->length), + ','); // NOLINT() + } +#endif + + // Push data to ring buffer (now safe in main loop) + channel->input_buffer_.push(chunk->data, chunk->length); + + // Return chunk to pool for reuse + this->chunk_pool_.release(chunk); + } + + // Log dropped USB data periodically + uint16_t dropped = this->usb_data_queue_.get_and_reset_dropped_count(); + if (dropped > 0) { + ESP_LOGW(TAG, "Dropped %u USB data chunks due to buffer overflow", dropped); + } +} void USBUartComponent::dump_config() { USBClient::dump_config(); for (auto &channel : this->channels_) { @@ -187,49 +214,70 @@ void USBUartComponent::dump_config() { } } void USBUartComponent::start_input(USBUartChannel *channel) { - if (!channel->initialised_ || channel->input_started_ || - channel->input_buffer_.get_free_space() < channel->cdc_dev_.in_ep->wMaxPacketSize) + if (!channel->initialised_.load() || channel->input_started_.load()) return; + // Note: This function is called from both USB task and main loop, so we cannot + // directly check ring buffer space here. Backpressure is handled by the chunk pool: + // when exhausted, USB input stops until chunks are freed by the main loop const auto *ep = channel->cdc_dev_.in_ep; + // CALLBACK CONTEXT: This lambda is executed in USB task via transfer_callback auto callback = [this, channel](const usb_host::TransferStatus &status) { ESP_LOGV(TAG, "Transfer result: length: %u; status %X", status.data_len, status.error_code); if (!status.success) { ESP_LOGE(TAG, "Control transfer failed, status=%s", esp_err_to_name(status.error_code)); + // On failure, don't restart - let next read_array() trigger it + channel->input_started_.store(false); return; } -#ifdef USE_UART_DEBUGGER - if (channel->debug_) { - uart::UARTDebug::log_hex(uart::UART_DIRECTION_RX, - std::vector(status.data, status.data + status.data_len), ','); // NOLINT() - } -#endif - channel->input_started_ = false; - if (!channel->dummy_receiver_) { - for (size_t i = 0; i != status.data_len; i++) { - channel->input_buffer_.push(status.data[i]); + + if (!channel->dummy_receiver_ && status.data_len > 0) { + // Allocate a chunk from the pool + UsbDataChunk *chunk = this->chunk_pool_.allocate(); + if (chunk == nullptr) { + // No chunks available - queue is full or we're out of memory + this->usb_data_queue_.increment_dropped_count(); + // Mark input as not started so we can retry + channel->input_started_.store(false); + return; } + + // Copy data to chunk (this is fast, happens in USB task) + memcpy(chunk->data, status.data, status.data_len); + chunk->length = status.data_len; + chunk->channel = channel; + + // Push to lock-free queue for main loop processing + // Push always succeeds because pool size == queue size + this->usb_data_queue_.push(chunk); } - if (channel->input_buffer_.get_free_space() >= channel->cdc_dev_.in_ep->wMaxPacketSize) { - this->defer([this, channel] { this->start_input(channel); }); - } + + // On success, restart input immediately from USB task for performance + // The lock-free queue will handle backpressure + channel->input_started_.store(false); + this->start_input(channel); }; - channel->input_started_ = true; + channel->input_started_.store(true); this->transfer_in(ep->bEndpointAddress, callback, ep->wMaxPacketSize); } void USBUartComponent::start_output(USBUartChannel *channel) { - if (channel->output_started_) + // IMPORTANT: This function must only be called from the main loop! + // The output_buffer_ is not thread-safe and can only be accessed from main loop. + // USB callbacks use defer() to ensure this function runs in the correct context. + if (channel->output_started_.load()) return; if (channel->output_buffer_.is_empty()) { return; } const auto *ep = channel->cdc_dev_.out_ep; + // CALLBACK CONTEXT: This lambda is executed in USB task via transfer_callback auto callback = [this, channel](const usb_host::TransferStatus &status) { ESP_LOGV(TAG, "Output Transfer result: length: %u; status %X", status.data_len, status.error_code); - channel->output_started_ = false; + channel->output_started_.store(false); + // Defer restart to main loop (defer is thread-safe) this->defer([this, channel] { this->start_output(channel); }); }; - channel->output_started_ = true; + channel->output_started_.store(true); uint8_t data[ep->wMaxPacketSize]; auto len = channel->output_buffer_.pop(data, ep->wMaxPacketSize); this->transfer_out(ep->bEndpointAddress, callback, data, len); @@ -272,7 +320,7 @@ void USBUartTypeCdcAcm::on_connected() { channel->cdc_dev_ = cdc_devs[i++]; fix_mps(channel->cdc_dev_.in_ep); fix_mps(channel->cdc_dev_.out_ep); - channel->initialised_ = true; + channel->initialised_.store(true); auto err = usb_host_interface_claim(this->handle_, this->device_handle_, channel->cdc_dev_.bulk_interface_number, 0); if (err != ESP_OK) { @@ -301,9 +349,9 @@ void USBUartTypeCdcAcm::on_disconnected() { usb_host_endpoint_flush(this->device_handle_, channel->cdc_dev_.notify_ep->bEndpointAddress); } usb_host_interface_release(this->handle_, this->device_handle_, channel->cdc_dev_.bulk_interface_number); - channel->initialised_ = false; - channel->input_started_ = false; - channel->output_started_ = false; + channel->initialised_.store(false); + channel->input_started_.store(false); + channel->output_started_.store(false); channel->input_buffer_.clear(); channel->output_buffer_.clear(); } @@ -312,10 +360,10 @@ void USBUartTypeCdcAcm::on_disconnected() { void USBUartTypeCdcAcm::enable_channels() { for (auto *channel : this->channels_) { - if (!channel->initialised_) + if (!channel->initialised_.load()) continue; - channel->input_started_ = false; - channel->output_started_ = false; + channel->input_started_.store(false); + channel->output_started_.store(false); this->start_input(channel); } } diff --git a/esphome/components/usb_uart/usb_uart.h b/esphome/components/usb_uart/usb_uart.h index a103c51add..b41e0a52e9 100644 --- a/esphome/components/usb_uart/usb_uart.h +++ b/esphome/components/usb_uart/usb_uart.h @@ -5,11 +5,15 @@ #include "esphome/core/helpers.h" #include "esphome/components/uart/uart_component.h" #include "esphome/components/usb_host/usb_host.h" +#include "esphome/core/lock_free_queue.h" +#include "esphome/core/event_pool.h" +#include namespace esphome { namespace usb_uart { class USBUartTypeCdcAcm; class USBUartComponent; +class USBUartChannel; static const char *const TAG = "usb_uart"; @@ -68,6 +72,17 @@ class RingBuffer { uint8_t *buffer_; }; +// Structure for queuing received USB data chunks +struct UsbDataChunk { + static constexpr size_t MAX_CHUNK_SIZE = 64; // USB packet size + uint8_t data[MAX_CHUNK_SIZE]; + uint8_t length; // Max 64 bytes, so uint8_t is sufficient + USBUartChannel *channel; + + // Required for EventPool - no cleanup needed for POD types + void release() {} +}; + class USBUartChannel : public uart::UARTComponent, public Parented { friend class USBUartComponent; friend class USBUartTypeCdcAcm; @@ -90,16 +105,20 @@ class USBUartChannel : public uart::UARTComponent, public Parenteddummy_receiver_ = dummy_receiver; } protected: - const uint8_t index_; + // Larger structures first for better alignment RingBuffer input_buffer_; RingBuffer output_buffer_; - UARTParityOptions parity_{UART_CONFIG_PARITY_NONE}; - bool input_started_{true}; - bool output_started_{true}; CdcEps cdc_dev_{}; + // Enum (likely 4 bytes) + UARTParityOptions parity_{UART_CONFIG_PARITY_NONE}; + // Group atomics together (each 1 byte) + std::atomic input_started_{true}; + std::atomic output_started_{true}; + std::atomic initialised_{false}; + // Group regular bytes together to minimize padding + const uint8_t index_; bool debug_{}; bool dummy_receiver_{}; - bool initialised_{}; }; class USBUartComponent : public usb_host::USBClient { @@ -115,6 +134,11 @@ class USBUartComponent : public usb_host::USBClient { void start_input(USBUartChannel *channel); void start_output(USBUartChannel *channel); + // Lock-free data transfer from USB task to main loop + static constexpr int USB_DATA_QUEUE_SIZE = 32; + LockFreeQueue usb_data_queue_; + EventPool chunk_pool_; + protected: std::vector channels_{}; };