Implement a lock free ring buffer for BLEScanResult to avoid drops (#9087)

This commit is contained in:
J. Nick Koston 2025-06-15 15:53:45 -05:00 committed by GitHub
parent 59f69ac5ca
commit 61a558a062
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 69 additions and 33 deletions

View File

@ -51,15 +51,14 @@ void ESP32BLETracker::setup() {
return; return;
} }
RAMAllocator<BLEScanResult> allocator; RAMAllocator<BLEScanResult> allocator;
this->scan_result_buffer_ = allocator.allocate(SCAN_RESULT_BUFFER_SIZE); this->scan_ring_buffer_ = allocator.allocate(SCAN_RESULT_BUFFER_SIZE);
if (this->scan_result_buffer_ == nullptr) { if (this->scan_ring_buffer_ == nullptr) {
ESP_LOGE(TAG, "Could not allocate buffer for BLE Tracker!"); ESP_LOGE(TAG, "Could not allocate ring buffer for BLE Tracker!");
this->mark_failed(); this->mark_failed();
} }
global_esp32_ble_tracker = this; global_esp32_ble_tracker = this;
this->scan_result_lock_ = xSemaphoreCreateMutex();
#ifdef USE_OTA #ifdef USE_OTA
ota::get_global_ota_callback()->add_on_state_callback( ota::get_global_ota_callback()->add_on_state_callback(
@ -119,27 +118,31 @@ void ESP32BLETracker::loop() {
} }
bool promote_to_connecting = discovered && !searching && !connecting; bool promote_to_connecting = discovered && !searching && !connecting;
if (this->scanner_state_ == ScannerState::RUNNING && // Process scan results from lock-free SPSC ring buffer
this->scan_result_index_ && // if it looks like we have a scan result we will take the lock // Consumer side: This runs in the main loop thread
xSemaphoreTake(this->scan_result_lock_, 0)) { if (this->scanner_state_ == ScannerState::RUNNING) {
uint32_t index = this->scan_result_index_; // Load our own index with relaxed ordering (we're the only writer)
if (index >= SCAN_RESULT_BUFFER_SIZE) { size_t read_idx = this->ring_read_index_.load(std::memory_order_relaxed);
ESP_LOGW(TAG, "Too many BLE events to process. Some devices may not show up.");
}
if (this->raw_advertisements_) { // Load producer's index with acquire to see their latest writes
for (auto *listener : this->listeners_) { size_t write_idx = this->ring_write_index_.load(std::memory_order_acquire);
listener->parse_devices(this->scan_result_buffer_, this->scan_result_index_);
}
for (auto *client : this->clients_) {
client->parse_devices(this->scan_result_buffer_, this->scan_result_index_);
}
}
if (this->parse_advertisements_) { while (read_idx != write_idx) {
for (size_t i = 0; i < index; i++) { // Process one result at a time directly from ring buffer
BLEScanResult &scan_result = this->scan_ring_buffer_[read_idx];
if (this->raw_advertisements_) {
for (auto *listener : this->listeners_) {
listener->parse_devices(&scan_result, 1);
}
for (auto *client : this->clients_) {
client->parse_devices(&scan_result, 1);
}
}
if (this->parse_advertisements_) {
ESPBTDevice device; ESPBTDevice device;
device.parse_scan_rst(this->scan_result_buffer_[i]); device.parse_scan_rst(scan_result);
bool found = false; bool found = false;
for (auto *listener : this->listeners_) { for (auto *listener : this->listeners_) {
@ -160,9 +163,19 @@ void ESP32BLETracker::loop() {
this->print_bt_device_info(device); this->print_bt_device_info(device);
} }
} }
// Move to next entry in ring buffer
read_idx = (read_idx + 1) % SCAN_RESULT_BUFFER_SIZE;
// Store with release to ensure reads complete before index update
this->ring_read_index_.store(read_idx, std::memory_order_release);
}
// Log dropped results periodically
size_t dropped = this->scan_results_dropped_.exchange(0, std::memory_order_relaxed);
if (dropped > 0) {
ESP_LOGW(TAG, "Dropped %zu BLE scan results due to buffer overflow", dropped);
} }
this->scan_result_index_ = 0;
xSemaphoreGive(this->scan_result_lock_);
} }
if (this->scanner_state_ == ScannerState::STOPPED) { if (this->scanner_state_ == ScannerState::STOPPED) {
this->end_of_scan_(); // Change state to IDLE this->end_of_scan_(); // Change state to IDLE
@ -391,12 +404,27 @@ void ESP32BLETracker::gap_scan_event_handler(const BLEScanResult &scan_result) {
ESP_LOGV(TAG, "gap_scan_result - event %d", scan_result.search_evt); ESP_LOGV(TAG, "gap_scan_result - event %d", scan_result.search_evt);
if (scan_result.search_evt == ESP_GAP_SEARCH_INQ_RES_EVT) { if (scan_result.search_evt == ESP_GAP_SEARCH_INQ_RES_EVT) {
if (xSemaphoreTake(this->scan_result_lock_, 0)) { // Lock-free SPSC ring buffer write (Producer side)
if (this->scan_result_index_ < SCAN_RESULT_BUFFER_SIZE) { // This runs in the ESP-IDF Bluetooth stack callback thread
// Store BLEScanResult directly in our buffer // IMPORTANT: Only this thread writes to ring_write_index_
this->scan_result_buffer_[this->scan_result_index_++] = scan_result;
} // Load our own index with relaxed ordering (we're the only writer)
xSemaphoreGive(this->scan_result_lock_); size_t write_idx = this->ring_write_index_.load(std::memory_order_relaxed);
size_t next_write_idx = (write_idx + 1) % SCAN_RESULT_BUFFER_SIZE;
// Load consumer's index with acquire to see their latest updates
size_t read_idx = this->ring_read_index_.load(std::memory_order_acquire);
// Check if buffer is full
if (next_write_idx != read_idx) {
// Write to ring buffer
this->scan_ring_buffer_[write_idx] = scan_result;
// Store with release to ensure the write is visible before index update
this->ring_write_index_.store(next_write_idx, std::memory_order_release);
} else {
// Buffer full, track dropped results
this->scan_results_dropped_.fetch_add(1, std::memory_order_relaxed);
} }
} else if (scan_result.search_evt == ESP_GAP_SEARCH_INQ_CMPL_EVT) { } else if (scan_result.search_evt == ESP_GAP_SEARCH_INQ_CMPL_EVT) {
// Scan finished on its own // Scan finished on its own

View File

@ -6,6 +6,7 @@
#include "esphome/core/helpers.h" #include "esphome/core/helpers.h"
#include <array> #include <array>
#include <atomic>
#include <string> #include <string>
#include <vector> #include <vector>
@ -282,9 +283,16 @@ class ESP32BLETracker : public Component,
bool ble_was_disabled_{true}; bool ble_was_disabled_{true};
bool raw_advertisements_{false}; bool raw_advertisements_{false};
bool parse_advertisements_{false}; bool parse_advertisements_{false};
SemaphoreHandle_t scan_result_lock_;
size_t scan_result_index_{0}; // Lock-free Single-Producer Single-Consumer (SPSC) ring buffer for scan results
BLEScanResult *scan_result_buffer_; // Producer: ESP-IDF Bluetooth stack callback (gap_scan_event_handler)
// Consumer: ESPHome main loop (loop() method)
// This design ensures zero blocking in the BT callback and prevents scan result loss
BLEScanResult *scan_ring_buffer_;
std::atomic<size_t> ring_write_index_{0}; // Written only by BT callback (producer)
std::atomic<size_t> ring_read_index_{0}; // Written only by main loop (consumer)
std::atomic<size_t> scan_results_dropped_{0}; // Tracks buffer overflow events
esp_bt_status_t scan_start_failed_{ESP_BT_STATUS_SUCCESS}; esp_bt_status_t scan_start_failed_{ESP_BT_STATUS_SUCCESS};
esp_bt_status_t scan_set_param_failed_{ESP_BT_STATUS_SUCCESS}; esp_bt_status_t scan_set_param_failed_{ESP_BT_STATUS_SUCCESS};
int connecting_{0}; int connecting_{0};