Merge branch 'usb_host_blocking_fix' into integration

esphome/components/usb_host/usb_host.h

@@ -5,7 +5,10 @@
 #include "esphome/core/component.h"
 #include <vector>
 #include "usb/usb_host.h"
-
+#include <freertos/FreeRTOS.h>
+#include <freertos/task.h>
+#include "esphome/core/lock_free_queue.h"
+#include "esphome/core/event_pool.h"
 #include <list>
 
 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<uint8_t> &data = {});
 
+  // Lock-free event queue and pool for USB task to main loop communication
+  // Must be public for access from static callbacks
+  LockFreeQueue<UsbEvent, USB_EVENT_QUEUE_SIZE> event_queue;
+  EventPool<UsbEvent, USB_EVENT_QUEUE_SIZE> 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};

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<USBClient *>(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");
-      ESP_LOGD(TAG, "Device gone %d", event_msg->new_dev.address);
+      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<USBClient *>(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<TransferRequest *>(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<TransferRequest *>(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.

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<uint8_t>(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)))

esphome/components/usb_uart/cp210x.cpp

@@ -100,12 +100,12 @@ std::vector<CdcEps> 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);

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<uint8_t>(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_ ||
+  if (!channel->initialised_.load() || channel->input_started_.load())
-      channel->input_buffer_.get_free_space() < channel->cdc_dev_.in_ep->wMaxPacketSize)
     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_) {
+    if (!channel->dummy_receiver_ && status.data_len > 0) {
-      uart::UARTDebug::log_hex(uart::UART_DIRECTION_RX,
+      // Allocate a chunk from the pool
-                               std::vector<uint8_t>(status.data, status.data + status.data_len), ',');  // NOLINT()
+      UsbDataChunk *chunk = this->chunk_pool_.allocate();
-    }
+      if (chunk == nullptr) {
-#endif
+        // No chunks available - queue is full or we're out of memory
-    channel->input_started_ = false;
+        this->usb_data_queue_.increment_dropped_count();
-    if (!channel->dummy_receiver_) {
+        // Mark input as not started so we can retry
-      for (size_t i = 0; i != status.data_len; i++) {
+        channel->input_started_.store(false);
-        channel->input_buffer_.push(status.data[i]);
+        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->initialised_.store(false);
-    channel->input_started_ = false;
+    channel->input_started_.store(false);
-    channel->output_started_ = 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->input_started_.store(false);
-    channel->output_started_ = false;
+    channel->output_started_.store(false);
     this->start_input(channel);
   }
 }

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 <atomic>
 
 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<USBUartComponent> {
   friend class USBUartComponent;
   friend class USBUartTypeCdcAcm;
@@ -90,16 +105,20 @@ class USBUartChannel : public uart::UARTComponent, public Parented<USBUartCompon
   void set_dummy_receiver(bool dummy_receiver) { this->dummy_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<bool> input_started_{true};
+  std::atomic<bool> output_started_{true};
+  std::atomic<bool> 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<UsbDataChunk, USB_DATA_QUEUE_SIZE> usb_data_queue_;
+  EventPool<UsbDataChunk, USB_DATA_QUEUE_SIZE> chunk_pool_;
+
  protected:
   std::vector<USBUartChannel *> channels_{};
 };