Extract lock-free queue and event pool to core helpers (#9238)

esphome/components/esp32_ble/ble.cpp

@@ -1,7 +1,6 @@
 #ifdef USE_ESP32
 
 #include "ble.h"
-#include "ble_event_pool.h"
 
 #include "esphome/core/application.h"
 #include "esphome/core/helpers.h"

esphome/components/esp32_ble/ble.h

@@ -12,8 +12,8 @@
 #include "esphome/core/helpers.h"
 
 #include "ble_event.h"
-#include "ble_event_pool.h"
-#include "queue.h"
+#include "esphome/core/lock_free_queue.h"
+#include "esphome/core/event_pool.h"
 
 #ifdef USE_ESP32
 
@@ -148,8 +148,8 @@ class ESP32BLE : public Component {
   std::vector<BLEStatusEventHandler *> ble_status_event_handlers_;
   BLEComponentState state_{BLE_COMPONENT_STATE_OFF};
 
-  LockFreeQueue<BLEEvent, MAX_BLE_QUEUE_SIZE> ble_events_;
-  BLEEventPool<MAX_BLE_QUEUE_SIZE> ble_event_pool_;
+  esphome::LockFreeQueue<BLEEvent, MAX_BLE_QUEUE_SIZE> ble_events_;
+  esphome::EventPool<BLEEvent, MAX_BLE_QUEUE_SIZE> ble_event_pool_;
   BLEAdvertising *advertising_{};
   esp_ble_io_cap_t io_cap_{ESP_IO_CAP_NONE};
   uint32_t advertising_cycle_time_{};

esphome/components/esp32_ble/ble_event.h

@@ -134,13 +134,13 @@ class BLEEvent {
   }
 
   // Destructor to clean up heap allocations
-  ~BLEEvent() { this->cleanup_heap_data(); }
+  ~BLEEvent() { this->release(); }
 
   // Default constructor for pre-allocation in pool
   BLEEvent() : type_(GAP) {}
 
-  // Clean up any heap-allocated data
-  void cleanup_heap_data() {
+  // Invoked on return to EventPool - clean up any heap-allocated data
+  void release() {
     if (this->type_ == GAP) {
       return;
     }
@@ -161,19 +161,19 @@ class BLEEvent {
 
   // Load new event data for reuse (replaces previous event data)
   void load_gap_event(esp_gap_ble_cb_event_t e, esp_ble_gap_cb_param_t *p) {
-    this->cleanup_heap_data();
+    this->release();
     this->type_ = GAP;
     this->init_gap_data_(e, p);
   }
 
   void load_gattc_event(esp_gattc_cb_event_t e, esp_gatt_if_t i, esp_ble_gattc_cb_param_t *p) {
-    this->cleanup_heap_data();
+    this->release();
     this->type_ = GATTC;
     this->init_gattc_data_(e, i, p);
   }
 
   void load_gatts_event(esp_gatts_cb_event_t e, esp_gatt_if_t i, esp_ble_gatts_cb_param_t *p) {
-    this->cleanup_heap_data();
+    this->release();
     this->type_ = GATTS;
     this->init_gatts_data_(e, i, p);
   }

esphome/components/esp32_ble/ble_event_pool.h

@@ -1,72 +0,0 @@
-#pragma once
-
-#ifdef USE_ESP32
-
-#include <atomic>
-#include <cstddef>
-#include "ble_event.h"
-#include "queue.h"
-#include "esphome/core/helpers.h"
-
-namespace esphome {
-namespace esp32_ble {
-
-// BLE Event Pool - On-demand pool of BLEEvent objects to avoid heap fragmentation
-// Events are allocated on first use and reused thereafter, growing to peak usage
-template<uint8_t SIZE> class BLEEventPool {
- public:
-  BLEEventPool() : total_created_(0) {}
-
-  ~BLEEventPool() {
-    // Clean up any remaining events in the free list
-    BLEEvent *event;
-    while ((event = this->free_list_.pop()) != nullptr) {
-      delete event;
-    }
-  }
-
-  // Allocate an event from the pool
-  // Returns nullptr if pool is full
-  BLEEvent *allocate() {
-    // Try to get from free list first
-    BLEEvent *event = this->free_list_.pop();
-    if (event != nullptr)
-      return event;
-
-    // Need to create a new event
-    if (this->total_created_ >= SIZE) {
-      // Pool is at capacity
-      return nullptr;
-    }
-
-    // Use internal RAM for better performance
-    RAMAllocator<BLEEvent> allocator(RAMAllocator<BLEEvent>::ALLOC_INTERNAL);
-    event = allocator.allocate(1);
-
-    if (event == nullptr) {
-      // Memory allocation failed
-      return nullptr;
-    }
-
-    // Placement new to construct the object
-    new (event) BLEEvent();
-    this->total_created_++;
-    return event;
-  }
-
-  // Return an event to the pool for reuse
-  void release(BLEEvent *event) {
-    if (event != nullptr) {
-      this->free_list_.push(event);
-    }
-  }
-
- private:
-  LockFreeQueue<BLEEvent, SIZE> free_list_;  // Free events ready for reuse
-  uint8_t total_created_;                    // Total events created (high water mark)
-};
-
-}  // namespace esp32_ble
-}  // namespace esphome
-
-#endif

esphome/components/esp32_ble/queue.h

@@ -1,85 +0,0 @@
-#pragma once
-
-#ifdef USE_ESP32
-
-#include <atomic>
-#include <cstddef>
-
-/*
- * BLE events come in from a separate Task (thread) in the ESP32 stack. Rather
- * than using mutex-based locking, this lock-free queue allows the BLE
- * task to enqueue events without blocking. The main loop() then processes
- * these events at a safer time.
- *
- * This is a Single-Producer Single-Consumer (SPSC) lock-free ring buffer.
- * The BLE task is the only producer, and the main loop() is the only consumer.
- */
-
-namespace esphome {
-namespace esp32_ble {
-
-template<class T, uint8_t SIZE> class LockFreeQueue {
- public:
-  LockFreeQueue() : head_(0), tail_(0), dropped_count_(0) {}
-
-  bool push(T *element) {
-    if (element == nullptr)
-      return false;
-
-    uint8_t current_tail = tail_.load(std::memory_order_relaxed);
-    uint8_t next_tail = (current_tail + 1) % SIZE;
-
-    if (next_tail == head_.load(std::memory_order_acquire)) {
-      // Buffer full
-      dropped_count_.fetch_add(1, std::memory_order_relaxed);
-      return false;
-    }
-
-    buffer_[current_tail] = element;
-    tail_.store(next_tail, std::memory_order_release);
-    return true;
-  }
-
-  T *pop() {
-    uint8_t current_head = head_.load(std::memory_order_relaxed);
-
-    if (current_head == tail_.load(std::memory_order_acquire)) {
-      return nullptr;  // Empty
-    }
-
-    T *element = buffer_[current_head];
-    head_.store((current_head + 1) % SIZE, std::memory_order_release);
-    return element;
-  }
-
-  size_t size() const {
-    uint8_t tail = tail_.load(std::memory_order_acquire);
-    uint8_t head = head_.load(std::memory_order_acquire);
-    return (tail - head + SIZE) % SIZE;
-  }
-
-  uint16_t get_and_reset_dropped_count() { return dropped_count_.exchange(0, std::memory_order_relaxed); }
-
-  void increment_dropped_count() { dropped_count_.fetch_add(1, std::memory_order_relaxed); }
-
-  bool empty() const { return head_.load(std::memory_order_acquire) == tail_.load(std::memory_order_acquire); }
-
-  bool full() const {
-    uint8_t next_tail = (tail_.load(std::memory_order_relaxed) + 1) % SIZE;
-    return next_tail == head_.load(std::memory_order_acquire);
-  }
-
- protected:
-  T *buffer_[SIZE];
-  // Atomic: written by producer (push/increment), read+reset by consumer (get_and_reset)
-  std::atomic<uint16_t> dropped_count_;  // 65535 max - more than enough for drop tracking
-  // Atomic: written by consumer (pop), read by producer (push) to check if full
-  std::atomic<uint8_t> head_;
-  // Atomic: written by producer (push), read by consumer (pop) to check if empty
-  std::atomic<uint8_t> tail_;
-};
-
-}  // namespace esp32_ble
-}  // namespace esphome
-
-#endif

esphome/core/event_pool.h

@@ -0,0 +1,81 @@
+#pragma once
+
+#if defined(USE_ESP32) || defined(USE_LIBRETINY)
+
+#include <atomic>
+#include <cstddef>
+#include "esphome/core/helpers.h"
+#include "esphome/core/lock_free_queue.h"
+
+namespace esphome {
+
+// Event Pool - On-demand pool of objects to avoid heap fragmentation
+// Events are allocated on first use and reused thereafter, growing to peak usage
+// @tparam T The type of objects managed by the pool (must have a release() method)
+// @tparam SIZE The maximum number of objects in the pool (1-255, limited by uint8_t)
+template<class T, uint8_t SIZE> class EventPool {
+ public:
+  EventPool() : total_created_(0) {}
+
+  ~EventPool() {
+    // Clean up any remaining events in the free list
+    // IMPORTANT: This destructor assumes no concurrent access. The EventPool must not
+    // be destroyed while any thread might still call allocate() or release().
+    // In practice, this is typically ensured by destroying the pool only during
+    // component shutdown when all producer/consumer threads have been stopped.
+    T *event;
+    RAMAllocator<T> allocator(RAMAllocator<T>::ALLOC_INTERNAL);
+    while ((event = this->free_list_.pop()) != nullptr) {
+      // Call destructor
+      event->~T();
+      // Deallocate using RAMAllocator
+      allocator.deallocate(event, 1);
+    }
+  }
+
+  // Allocate an event from the pool
+  // Returns nullptr if pool is full
+  T *allocate() {
+    // Try to get from free list first
+    T *event = this->free_list_.pop();
+    if (event != nullptr)
+      return event;
+
+    // Need to create a new event
+    if (this->total_created_ >= SIZE) {
+      // Pool is at capacity
+      return nullptr;
+    }
+
+    // Use internal RAM for better performance
+    RAMAllocator<T> allocator(RAMAllocator<T>::ALLOC_INTERNAL);
+    event = allocator.allocate(1);
+
+    if (event == nullptr) {
+      // Memory allocation failed
+      return nullptr;
+    }
+
+    // Placement new to construct the object
+    new (event) T();
+    this->total_created_++;
+    return event;
+  }
+
+  // Return an event to the pool for reuse
+  void release(T *event) {
+    if (event != nullptr) {
+      // Clean up the event's allocated memory
+      event->release();
+      this->free_list_.push(event);
+    }
+  }
+
+ private:
+  LockFreeQueue<T, SIZE> free_list_;  // Free events ready for reuse
+  uint8_t total_created_;             // Total events created (high water mark, max 255)
+};
+
+}  // namespace esphome
+
+#endif  // defined(USE_ESP32) || defined(USE_LIBRETINY)

esphome/core/lock_free_queue.h

@@ -0,0 +1,132 @@
+#pragma once
+
+#if defined(USE_ESP32) || defined(USE_LIBRETINY)
+
+#include <atomic>
+#include <cstddef>
+
+#if defined(USE_ESP32)
+#include <freertos/FreeRTOS.h>
+#include <freertos/task.h>
+#elif defined(USE_LIBRETINY)
+#include <FreeRTOS.h>
+#include <task.h>
+#endif
+
+/*
+ * Lock-free queue for single-producer single-consumer scenarios.
+ * This allows one thread to push items and another to pop them without
+ * blocking each other.
+ *
+ * This is a Single-Producer Single-Consumer (SPSC) lock-free ring buffer.
+ * Available on platforms with FreeRTOS support (ESP32, LibreTiny).
+ *
+ * Common use cases:
+ * - BLE events: BLE task produces, main loop consumes
+ * - MQTT messages: main task produces, MQTT thread consumes
+ *
+ * @tparam T The type of elements stored in the queue (must be a pointer type)
+ * @tparam SIZE The maximum number of elements (1-255, limited by uint8_t indices)
+ */
+
+namespace esphome {
+
+template<class T, uint8_t SIZE> class LockFreeQueue {
+ public:
+  LockFreeQueue() : head_(0), tail_(0), dropped_count_(0), task_to_notify_(nullptr) {}
+
+  bool push(T *element) {
+    if (element == nullptr)
+      return false;
+
+    uint8_t current_tail = tail_.load(std::memory_order_relaxed);
+    uint8_t next_tail = (current_tail + 1) % SIZE;
+
+    // Read head before incrementing tail
+    uint8_t head_before = head_.load(std::memory_order_acquire);
+
+    if (next_tail == head_before) {
+      // Buffer full
+      dropped_count_.fetch_add(1, std::memory_order_relaxed);
+      return false;
+    }
+
+    // Check if queue was empty before push
+    bool was_empty = (current_tail == head_before);
+
+    buffer_[current_tail] = element;
+    tail_.store(next_tail, std::memory_order_release);
+
+    // Notify optimization: only notify if we need to
+    if (task_to_notify_ != nullptr) {
+      if (was_empty) {
+        // Queue was empty - consumer might be going to sleep, must notify
+        xTaskNotifyGive(task_to_notify_);
+      } else {
+        // Queue wasn't empty - check if consumer has caught up to previous tail
+        uint8_t head_after = head_.load(std::memory_order_acquire);
+        if (head_after == current_tail) {
+          // Consumer just caught up to where tail was - might go to sleep, must notify
+          // Note: There's a benign race here - between reading head_after and calling
+          // xTaskNotifyGive(), the consumer could advance further. This would result
+          // in an unnecessary wake-up, but is harmless and extremely rare in practice.
+          xTaskNotifyGive(task_to_notify_);
+        }
+        // Otherwise: consumer is still behind, no need to notify
+      }
+    }
+
+    return true;
+  }
+
+  T *pop() {
+    uint8_t current_head = head_.load(std::memory_order_relaxed);
+
+    if (current_head == tail_.load(std::memory_order_acquire)) {
+      return nullptr;  // Empty
+    }
+
+    T *element = buffer_[current_head];
+    head_.store((current_head + 1) % SIZE, std::memory_order_release);
+    return element;
+  }
+
+  size_t size() const {
+    uint8_t tail = tail_.load(std::memory_order_acquire);
+    uint8_t head = head_.load(std::memory_order_acquire);
+    return (tail - head + SIZE) % SIZE;
+  }
+
+  uint16_t get_and_reset_dropped_count() { return dropped_count_.exchange(0, std::memory_order_relaxed); }
+
+  void increment_dropped_count() { dropped_count_.fetch_add(1, std::memory_order_relaxed); }
+
+  bool empty() const { return head_.load(std::memory_order_acquire) == tail_.load(std::memory_order_acquire); }
+
+  bool full() const {
+    uint8_t next_tail = (tail_.load(std::memory_order_relaxed) + 1) % SIZE;
+    return next_tail == head_.load(std::memory_order_acquire);
+  }
+
+  // Set the FreeRTOS task handle to notify when items are pushed to the queue
+  // This enables efficient wake-up of a consumer task that's waiting for data
+  // @param task The FreeRTOS task handle to notify, or nullptr to disable notifications
+  void set_task_to_notify(TaskHandle_t task) { task_to_notify_ = task; }
+
+ protected:
+  T *buffer_[SIZE];
+  // Atomic: written by producer (push/increment), read+reset by consumer (get_and_reset)
+  std::atomic<uint16_t> dropped_count_;  // 65535 max - more than enough for drop tracking
+  // Atomic: written by consumer (pop), read by producer (push) to check if full
+  // Using uint8_t limits queue size to 255 elements but saves memory and ensures
+  // atomic operations are efficient on all platforms
+  std::atomic<uint8_t> head_;
+  // Atomic: written by producer (push), read by consumer (pop) to check if empty
+  std::atomic<uint8_t> tail_;
+  // Task handle for notification (optional)
+  TaskHandle_t task_to_notify_;
+};
+
+}  // namespace esphome
+
+#endif  // defined(USE_ESP32) || defined(USE_LIBRETINY)