From 6e31fb181ee6bbe4c285332c1d31714547eb644c Mon Sep 17 00:00:00 2001 From: "@RubenKelevra" Date: Sun, 20 Jul 2025 23:57:52 +0200 Subject: [PATCH 1/4] core/scheduler: Make millis_64_ rollover monotonic on SMP (#9716) Co-authored-by: J. Nick Koston --- esphome/components/esp32/__init__.py | 2 + esphome/components/esp8266/__init__.py | 2 + esphome/components/host/__init__.py | 2 + esphome/components/libretiny/__init__.py | 2 + esphome/components/rp2040/__init__.py | 2 + esphome/const.py | 8 + esphome/core/defines.h | 3 + esphome/core/scheduler.cpp | 197 ++++++++++++++--------- esphome/core/scheduler.h | 44 +++-- 9 files changed, 175 insertions(+), 87 deletions(-) diff --git a/esphome/components/esp32/__init__.py b/esphome/components/esp32/__init__.py index c772a3438c..6ddb579733 100644 --- a/esphome/components/esp32/__init__.py +++ b/esphome/components/esp32/__init__.py @@ -31,6 +31,7 @@ from esphome.const import ( KEY_TARGET_FRAMEWORK, KEY_TARGET_PLATFORM, PLATFORM_ESP32, + CoreModel, __version__, ) from esphome.core import CORE, HexInt, TimePeriod @@ -713,6 +714,7 @@ async def to_code(config): cg.add_define("ESPHOME_BOARD", config[CONF_BOARD]) cg.add_build_flag(f"-DUSE_ESP32_VARIANT_{config[CONF_VARIANT]}") cg.add_define("ESPHOME_VARIANT", VARIANT_FRIENDLY[config[CONF_VARIANT]]) + cg.add_define(CoreModel.MULTI_ATOMICS) cg.add_platformio_option("lib_ldf_mode", "off") cg.add_platformio_option("lib_compat_mode", "strict") diff --git a/esphome/components/esp8266/__init__.py b/esphome/components/esp8266/__init__.py index 01b20bdcb1..d08d7121b7 100644 --- a/esphome/components/esp8266/__init__.py +++ b/esphome/components/esp8266/__init__.py @@ -15,6 +15,7 @@ from esphome.const import ( KEY_TARGET_FRAMEWORK, KEY_TARGET_PLATFORM, PLATFORM_ESP8266, + CoreModel, ) from esphome.core import CORE, coroutine_with_priority from esphome.helpers import copy_file_if_changed @@ -187,6 +188,7 @@ async def to_code(config): cg.set_cpp_standard("gnu++20") cg.add_define("ESPHOME_BOARD", config[CONF_BOARD]) cg.add_define("ESPHOME_VARIANT", "ESP8266") + cg.add_define(CoreModel.SINGLE) cg.add_platformio_option("extra_scripts", ["post:post_build.py"]) diff --git a/esphome/components/host/__init__.py b/esphome/components/host/__init__.py index a67d73fbb7..2d77f2f7ab 100644 --- a/esphome/components/host/__init__.py +++ b/esphome/components/host/__init__.py @@ -7,6 +7,7 @@ from esphome.const import ( KEY_TARGET_FRAMEWORK, KEY_TARGET_PLATFORM, PLATFORM_HOST, + CoreModel, ) from esphome.core import CORE @@ -43,6 +44,7 @@ async def to_code(config): cg.add_define("USE_ESPHOME_HOST_MAC_ADDRESS", config[CONF_MAC_ADDRESS].parts) cg.add_build_flag("-std=gnu++20") cg.add_define("ESPHOME_BOARD", "host") + cg.add_define(CoreModel.MULTI_ATOMICS) cg.add_platformio_option("platform", "platformio/native") cg.add_platformio_option("lib_ldf_mode", "off") cg.add_platformio_option("lib_compat_mode", "strict") diff --git a/esphome/components/libretiny/__init__.py b/esphome/components/libretiny/__init__.py index 17d5d46ffd..7f2a0bc0a5 100644 --- a/esphome/components/libretiny/__init__.py +++ b/esphome/components/libretiny/__init__.py @@ -20,6 +20,7 @@ from esphome.const import ( KEY_FRAMEWORK_VERSION, KEY_TARGET_FRAMEWORK, KEY_TARGET_PLATFORM, + CoreModel, __version__, ) from esphome.core import CORE @@ -260,6 +261,7 @@ async def component_to_code(config): cg.add_build_flag(f"-DUSE_LIBRETINY_VARIANT_{config[CONF_FAMILY]}") cg.add_define("ESPHOME_BOARD", config[CONF_BOARD]) cg.add_define("ESPHOME_VARIANT", FAMILY_FRIENDLY[config[CONF_FAMILY]]) + cg.add_define(CoreModel.MULTI_NO_ATOMICS) # force using arduino framework cg.add_platformio_option("framework", "arduino") diff --git a/esphome/components/rp2040/__init__.py b/esphome/components/rp2040/__init__.py index 0fa299ce5c..28c3bbd70c 100644 --- a/esphome/components/rp2040/__init__.py +++ b/esphome/components/rp2040/__init__.py @@ -16,6 +16,7 @@ from esphome.const import ( KEY_TARGET_FRAMEWORK, KEY_TARGET_PLATFORM, PLATFORM_RP2040, + CoreModel, ) from esphome.core import CORE, EsphomeError, coroutine_with_priority from esphome.helpers import copy_file_if_changed, mkdir_p, read_file, write_file @@ -171,6 +172,7 @@ async def to_code(config): cg.set_cpp_standard("gnu++20") cg.add_define("ESPHOME_BOARD", config[CONF_BOARD]) cg.add_define("ESPHOME_VARIANT", "RP2040") + cg.add_define(CoreModel.SINGLE) cg.add_platformio_option("extra_scripts", ["post:post_build.py"]) diff --git a/esphome/const.py b/esphome/const.py index 7da19a8c1b..627b6bac18 100644 --- a/esphome/const.py +++ b/esphome/const.py @@ -35,6 +35,14 @@ class Framework(StrEnum): ZEPHYR = "zephyr" +class CoreModel(StrEnum): + """Core model identifiers for ESPHome scheduler.""" + + SINGLE = "ESPHOME_CORES_SINGLE" + MULTI_NO_ATOMICS = "ESPHOME_CORES_MULTI_NO_ATOMICS" + MULTI_ATOMICS = "ESPHOME_CORES_MULTI_ATOMICS" + + class PlatformFramework(Enum): """Combined platform-framework identifiers with tuple values.""" diff --git a/esphome/core/defines.h b/esphome/core/defines.h index 7ddb3436cd..19d380bd29 100644 --- a/esphome/core/defines.h +++ b/esphome/core/defines.h @@ -15,6 +15,9 @@ #define ESPHOME_VARIANT "ESP32" #define ESPHOME_DEBUG_SCHEDULER +// Default threading model for static analysis (ESP32 is multi-core with atomics) +#define ESPHOME_CORES_MULTI_ATOMICS + // logger #define ESPHOME_LOG_LEVEL ESPHOME_LOG_LEVEL_VERY_VERBOSE diff --git a/esphome/core/scheduler.cpp b/esphome/core/scheduler.cpp index 7a0c08e1f0..d6d99f82c8 100644 --- a/esphome/core/scheduler.cpp +++ b/esphome/core/scheduler.cpp @@ -54,7 +54,7 @@ static void validate_static_string(const char *name) { ESP_LOGW(TAG, "WARNING: Scheduler name '%s' at %p might be on heap (static ref at %p)", name, name, static_str); } } -#endif +#endif /* ESPHOME_DEBUG_SCHEDULER */ // A note on locking: the `lock_` lock protects the `items_` and `to_add_` containers. It must be taken when writing to // them (i.e. when adding/removing items, but not when changing items). As items are only deleted from the loop task, @@ -82,9 +82,9 @@ void HOT Scheduler::set_timer_common_(Component *component, SchedulerItem::Type item->callback = std::move(func); item->remove = false; -#if !defined(USE_ESP8266) && !defined(USE_RP2040) +#ifndef ESPHOME_CORES_SINGLE // Special handling for defer() (delay = 0, type = TIMEOUT) - // ESP8266 and RP2040 are excluded because they don't need thread-safe defer handling + // Single-core platforms don't need thread-safe defer handling if (delay == 0 && type == SchedulerItem::TIMEOUT) { // Put in defer queue for guaranteed FIFO execution LockGuard guard{this->lock_}; @@ -92,7 +92,7 @@ void HOT Scheduler::set_timer_common_(Component *component, SchedulerItem::Type this->defer_queue_.push_back(std::move(item)); return; } -#endif +#endif /* not ESPHOME_CORES_SINGLE */ // Get fresh timestamp for new timer/interval - ensures accurate scheduling const auto now = this->millis_64_(millis()); // Fresh millis() call @@ -123,7 +123,7 @@ void HOT Scheduler::set_timer_common_(Component *component, SchedulerItem::Type ESP_LOGD(TAG, "set_%s(name='%s/%s', %s=%" PRIu32 ", offset=%" PRIu32 ")", type_str, item->get_source(), name_cstr ? name_cstr : "(null)", type_str, delay, static_cast(item->next_execution_ - now)); } -#endif +#endif /* ESPHOME_DEBUG_SCHEDULER */ LockGuard guard{this->lock_}; // If name is provided, do atomic cancel-and-add @@ -231,7 +231,7 @@ optional HOT Scheduler::next_schedule_in(uint32_t now) { return item->next_execution_ - now_64; } void HOT Scheduler::call(uint32_t now) { -#if !defined(USE_ESP8266) && !defined(USE_RP2040) +#ifndef ESPHOME_CORES_SINGLE // Process defer queue first to guarantee FIFO execution order for deferred items. // Previously, defer() used the heap which gave undefined order for equal timestamps, // causing race conditions on multi-core systems (ESP32, BK7200). @@ -239,8 +239,7 @@ void HOT Scheduler::call(uint32_t now) { // - Deferred items (delay=0) go directly to defer_queue_ in set_timer_common_ // - Items execute in exact order they were deferred (FIFO guarantee) // - No deferred items exist in to_add_, so processing order doesn't affect correctness - // ESP8266 and RP2040 don't use this queue - they fall back to the heap-based approach - // (ESP8266: single-core, RP2040: empty mutex implementation). + // Single-core platforms don't use this queue and fall back to the heap-based approach. // // Note: Items cancelled via cancel_item_locked_() are marked with remove=true but still // processed here. They are removed from the queue normally via pop_front() but skipped @@ -262,7 +261,7 @@ void HOT Scheduler::call(uint32_t now) { this->execute_item_(item.get(), now); } } -#endif +#endif /* not ESPHOME_CORES_SINGLE */ // Convert the fresh timestamp from main loop to 64-bit for scheduler operations const auto now_64 = this->millis_64_(now); // 'now' from parameter - fresh from Application::loop() @@ -274,13 +273,15 @@ void HOT Scheduler::call(uint32_t now) { if (now_64 - last_print > 2000) { last_print = now_64; std::vector> old_items; -#if !defined(USE_ESP8266) && !defined(USE_RP2040) && !defined(USE_LIBRETINY) - ESP_LOGD(TAG, "Items: count=%zu, now=%" PRIu64 " (%u, %" PRIu32 ")", this->items_.size(), now_64, - this->millis_major_, this->last_millis_.load(std::memory_order_relaxed)); -#else - ESP_LOGD(TAG, "Items: count=%zu, now=%" PRIu64 " (%u, %" PRIu32 ")", this->items_.size(), now_64, +#ifdef ESPHOME_CORES_MULTI_ATOMICS + const auto last_dbg = this->last_millis_.load(std::memory_order_relaxed); + const auto major_dbg = this->millis_major_.load(std::memory_order_relaxed); + ESP_LOGD(TAG, "Items: count=%zu, now=%" PRIu64 " (%" PRIu16 ", %" PRIu32 ")", this->items_.size(), now_64, + major_dbg, last_dbg); +#else /* not ESPHOME_CORES_MULTI_ATOMICS */ + ESP_LOGD(TAG, "Items: count=%zu, now=%" PRIu64 " (%" PRIu16 ", %" PRIu32 ")", this->items_.size(), now_64, this->millis_major_, this->last_millis_); -#endif +#endif /* else ESPHOME_CORES_MULTI_ATOMICS */ while (!this->empty_()) { std::unique_ptr item; { @@ -305,7 +306,7 @@ void HOT Scheduler::call(uint32_t now) { std::make_heap(this->items_.begin(), this->items_.end(), SchedulerItem::cmp); } } -#endif // ESPHOME_DEBUG_SCHEDULER +#endif /* ESPHOME_DEBUG_SCHEDULER */ // If we have too many items to remove if (this->to_remove_ > MAX_LOGICALLY_DELETED_ITEMS) { @@ -352,7 +353,7 @@ void HOT Scheduler::call(uint32_t now) { ESP_LOGV(TAG, "Running %s '%s/%s' with interval=%" PRIu32 " next_execution=%" PRIu64 " (now=%" PRIu64 ")", item->get_type_str(), item->get_source(), item_name ? item_name : "(null)", item->interval, item->next_execution_, now_64); -#endif +#endif /* ESPHOME_DEBUG_SCHEDULER */ // Warning: During callback(), a lot of stuff can happen, including: // - timeouts/intervals get added, potentially invalidating vector pointers @@ -460,7 +461,7 @@ bool HOT Scheduler::cancel_item_locked_(Component *component, const char *name_c size_t total_cancelled = 0; // Check all containers for matching items -#if !defined(USE_ESP8266) && !defined(USE_RP2040) +#ifndef ESPHOME_CORES_SINGLE // Only check defer queue for timeouts (intervals never go there) if (type == SchedulerItem::TIMEOUT) { for (auto &item : this->defer_queue_) { @@ -470,7 +471,7 @@ bool HOT Scheduler::cancel_item_locked_(Component *component, const char *name_c } } } -#endif +#endif /* not ESPHOME_CORES_SINGLE */ // Cancel items in the main heap for (auto &item : this->items_) { @@ -495,24 +496,53 @@ bool HOT Scheduler::cancel_item_locked_(Component *component, const char *name_c uint64_t Scheduler::millis_64_(uint32_t now) { // THREAD SAFETY NOTE: - // This function can be called from multiple threads simultaneously on ESP32/LibreTiny. - // On single-threaded platforms (ESP8266, RP2040), atomics are not needed. + // This function has three implementations, based on the precompiler flags + // - ESPHOME_CORES_SINGLE - Runs on single-core platforms (ESP8266, RP2040, etc.) + // - ESPHOME_CORES_MULTI_NO_ATOMICS - Runs on multi-core platforms without atomics (LibreTiny) + // - ESPHOME_CORES_MULTI_ATOMICS - Runs on multi-core platforms with atomics (ESP32, HOST, etc.) + // + // Make sure all changes are synchronized if you edit this function. // // IMPORTANT: Always pass fresh millis() values to this function. The implementation // handles out-of-order timestamps between threads, but minimizing time differences // helps maintain accuracy. // - // The implementation handles the 32-bit rollover (every 49.7 days) by: - // 1. Using a lock when detecting rollover to ensure atomic update - // 2. Restricting normal updates to forward movement within the same epoch - // This prevents race conditions at the rollover boundary without requiring - // 64-bit atomics or locking on every call. -#ifdef USE_LIBRETINY - // LibreTiny: Multi-threaded but lacks atomic operation support - // TODO: If LibreTiny ever adds atomic support, remove this entire block and - // let it fall through to the atomic-based implementation below - // We need to use a lock when near the rollover boundary to prevent races +#ifdef ESPHOME_CORES_SINGLE + // This is the single core implementation. + // + // Single-core platforms have no concurrency, so this is a simple implementation + // that just tracks 32-bit rollover (every 49.7 days) without any locking or atomics. + + uint16_t major = this->millis_major_; + uint32_t last = this->last_millis_; + + // Check for rollover + if (now < last && (last - now) > HALF_MAX_UINT32) { + this->millis_major_++; + major++; +#ifdef ESPHOME_DEBUG_SCHEDULER + ESP_LOGD(TAG, "Detected true 32-bit rollover at %" PRIu32 "ms (was %" PRIu32 ")", now, last); +#endif /* ESPHOME_DEBUG_SCHEDULER */ + } + + // Only update if time moved forward + if (now > last) { + this->last_millis_ = now; + } + + // Combine major (high 32 bits) and now (low 32 bits) into 64-bit time + return now + (static_cast(major) << 32); + +#elif defined(ESPHOME_CORES_MULTI_NO_ATOMICS) + // This is the multi core no atomics implementation. + // + // Without atomics, this implementation uses locks more aggressively: + // 1. Always locks when near the rollover boundary (within 10 seconds) + // 2. Always locks when detecting a large backwards jump + // 3. Updates without lock in normal forward progression (accepting minor races) + // This is less efficient but necessary without atomic operations. + uint16_t major = this->millis_major_; uint32_t last = this->last_millis_; // Define a safe window around the rollover point (10 seconds) @@ -531,9 +561,10 @@ uint64_t Scheduler::millis_64_(uint32_t now) { if (now < last && (last - now) > HALF_MAX_UINT32) { // True rollover detected (happens every ~49.7 days) this->millis_major_++; + major++; #ifdef ESPHOME_DEBUG_SCHEDULER ESP_LOGD(TAG, "Detected true 32-bit rollover at %" PRIu32 "ms (was %" PRIu32 ")", now, last); -#endif +#endif /* ESPHOME_DEBUG_SCHEDULER */ } // Update last_millis_ while holding lock this->last_millis_ = now; @@ -549,58 +580,76 @@ uint64_t Scheduler::millis_64_(uint32_t now) { // If now <= last and we're not near rollover, don't update // This minimizes backwards time movement -#elif !defined(USE_ESP8266) && !defined(USE_RP2040) - // Multi-threaded platforms with atomic support (ESP32) - uint32_t last = this->last_millis_.load(std::memory_order_relaxed); + // Combine major (high 32 bits) and now (low 32 bits) into 64-bit time + return now + (static_cast(major) << 32); - // If we might be near a rollover (large backwards jump), take the lock for the entire operation - // This ensures rollover detection and last_millis_ update are atomic together - if (now < last && (last - now) > HALF_MAX_UINT32) { - // Potential rollover - need lock for atomic rollover detection + update - LockGuard guard{this->lock_}; - // Re-read with lock held - last = this->last_millis_.load(std::memory_order_relaxed); +#elif defined(ESPHOME_CORES_MULTI_ATOMICS) + // This is the multi core with atomics implementation. + // + // Uses atomic operations with acquire/release semantics to ensure coherent + // reads of millis_major_ and last_millis_ across cores. Features: + // 1. Epoch-coherency retry loop to handle concurrent updates + // 2. Lock only taken for actual rollover detection and update + // 3. Lock-free CAS updates for normal forward time progression + // 4. Memory ordering ensures cores see consistent time values + for (;;) { + uint16_t major = this->millis_major_.load(std::memory_order_acquire); + + /* + * Acquire so that if we later decide **not** to take the lock we still + * observe a `millis_major_` value coherent with the loaded `last_millis_`. + * The acquire load ensures any later read of `millis_major_` sees its + * corresponding increment. + */ + uint32_t last = this->last_millis_.load(std::memory_order_acquire); + + // If we might be near a rollover (large backwards jump), take the lock for the entire operation + // This ensures rollover detection and last_millis_ update are atomic together if (now < last && (last - now) > HALF_MAX_UINT32) { - // True rollover detected (happens every ~49.7 days) - this->millis_major_++; + // Potential rollover - need lock for atomic rollover detection + update + LockGuard guard{this->lock_}; + // Re-read with lock held; mutex already provides ordering + last = this->last_millis_.load(std::memory_order_relaxed); + + if (now < last && (last - now) > HALF_MAX_UINT32) { + // True rollover detected (happens every ~49.7 days) + this->millis_major_.fetch_add(1, std::memory_order_relaxed); + major++; #ifdef ESPHOME_DEBUG_SCHEDULER - ESP_LOGD(TAG, "Detected true 32-bit rollover at %" PRIu32 "ms (was %" PRIu32 ")", now, last); -#endif - } - // Update last_millis_ while holding lock to prevent races - this->last_millis_.store(now, std::memory_order_relaxed); - } else { - // Normal case: Try lock-free update, but only allow forward movement within same epoch - // This prevents accidentally moving backwards across a rollover boundary - while (now > last && (now - last) < HALF_MAX_UINT32) { - if (this->last_millis_.compare_exchange_weak(last, now, std::memory_order_relaxed)) { - break; + ESP_LOGD(TAG, "Detected true 32-bit rollover at %" PRIu32 "ms (was %" PRIu32 ")", now, last); +#endif /* ESPHOME_DEBUG_SCHEDULER */ + } + /* + * Update last_millis_ while holding the lock to prevent races + * Publish the new low-word *after* bumping `millis_major_` (done above) + * so readers never see a mismatched pair. + */ + this->last_millis_.store(now, std::memory_order_release); + } else { + // Normal case: Try lock-free update, but only allow forward movement within same epoch + // This prevents accidentally moving backwards across a rollover boundary + while (now > last && (now - last) < HALF_MAX_UINT32) { + if (this->last_millis_.compare_exchange_weak(last, now, + std::memory_order_release, // success + std::memory_order_relaxed)) { // failure + break; + } + // CAS failure means no data was published; relaxed is fine + // last is automatically updated by compare_exchange_weak if it fails } - // last is automatically updated by compare_exchange_weak if it fails } + uint16_t major_end = this->millis_major_.load(std::memory_order_relaxed); + if (major_end == major) + return now + (static_cast(major) << 32); } + // Unreachable - the loop always returns when major_end == major + __builtin_unreachable(); #else - // Single-threaded platforms (ESP8266, RP2040): No atomics needed - uint32_t last = this->last_millis_; - - // Check for rollover - if (now < last && (last - now) > HALF_MAX_UINT32) { - this->millis_major_++; -#ifdef ESPHOME_DEBUG_SCHEDULER - ESP_LOGD(TAG, "Detected true 32-bit rollover at %" PRIu32 "ms (was %" PRIu32 ")", now, last); +#error \ + "No platform threading model defined. One of ESPHOME_CORES_SINGLE, ESPHOME_CORES_MULTI_NO_ATOMICS, or ESPHOME_CORES_MULTI_ATOMICS must be defined." #endif - } - - // Only update if time moved forward - if (now > last) { - this->last_millis_ = now; - } -#endif - - // Combine major (high 32 bits) and now (low 32 bits) into 64-bit time - return now + (static_cast(this->millis_major_) << 32); } bool HOT Scheduler::SchedulerItem::cmp(const std::unique_ptr &a, diff --git a/esphome/core/scheduler.h b/esphome/core/scheduler.h index 64df2f2bb0..b539b26949 100644 --- a/esphome/core/scheduler.h +++ b/esphome/core/scheduler.h @@ -1,10 +1,11 @@ #pragma once +#include "esphome/core/defines.h" #include #include #include #include -#if !defined(USE_ESP8266) && !defined(USE_RP2040) && !defined(USE_LIBRETINY) +#ifdef ESPHOME_CORES_MULTI_ATOMICS #include #endif @@ -204,23 +205,40 @@ class Scheduler { Mutex lock_; std::vector> items_; std::vector> to_add_; -#if !defined(USE_ESP8266) && !defined(USE_RP2040) - // ESP8266 and RP2040 don't need the defer queue because: - // ESP8266: Single-core with no preemptive multitasking - // RP2040: Currently has empty mutex implementation in ESPHome - // Both platforms save 40 bytes of RAM by excluding this +#ifndef ESPHOME_CORES_SINGLE + // Single-core platforms don't need the defer queue and save 40 bytes of RAM std::deque> defer_queue_; // FIFO queue for defer() calls -#endif -#if !defined(USE_ESP8266) && !defined(USE_RP2040) && !defined(USE_LIBRETINY) - // Multi-threaded platforms with atomic support: last_millis_ needs atomic for lock-free updates +#endif /* ESPHOME_CORES_SINGLE */ + uint32_t to_remove_{0}; + +#ifdef ESPHOME_CORES_MULTI_ATOMICS + /* + * Multi-threaded platforms with atomic support: last_millis_ needs atomic for lock-free updates + * + * MEMORY-ORDERING NOTE + * -------------------- + * `last_millis_` and `millis_major_` form a single 64-bit timestamp split in half. + * Writers publish `last_millis_` with memory_order_release and readers use + * memory_order_acquire. This ensures that once a reader sees the new low word, + * it also observes the corresponding increment of `millis_major_`. + */ std::atomic last_millis_{0}; -#else +#else /* not ESPHOME_CORES_MULTI_ATOMICS */ // Platforms without atomic support or single-threaded platforms uint32_t last_millis_{0}; -#endif - // millis_major_ is protected by lock when incrementing +#endif /* else ESPHOME_CORES_MULTI_ATOMICS */ + + /* + * Upper 16 bits of the 64-bit millis counter. Incremented only while holding + * `lock_`; read concurrently. Atomic (relaxed) avoids a formal data race. + * Ordering relative to `last_millis_` is provided by its release store and the + * corresponding acquire loads. + */ +#ifdef ESPHOME_CORES_MULTI_ATOMICS + std::atomic millis_major_{0}; +#else /* not ESPHOME_CORES_MULTI_ATOMICS */ uint16_t millis_major_{0}; - uint32_t to_remove_{0}; +#endif /* else ESPHOME_CORES_MULTI_ATOMICS */ }; } // namespace esphome From 335110d71fce3cc588d2af37b0681e7e9838c51f Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 20 Jul 2025 12:15:34 -1000 Subject: [PATCH 2/4] [bluetooth_proxy] Fix service discovery on disconnect and refactor connection handling (#9697) --- .../bluetooth_proxy/bluetooth_connection.cpp | 181 +++++++++++++++++- .../bluetooth_proxy/bluetooth_connection.h | 4 + .../bluetooth_proxy/bluetooth_proxy.cpp | 135 +------------ .../bluetooth_proxy/bluetooth_proxy.h | 6 +- 4 files changed, 183 insertions(+), 143 deletions(-) diff --git a/esphome/components/bluetooth_proxy/bluetooth_connection.cpp b/esphome/components/bluetooth_proxy/bluetooth_connection.cpp index 2bfccdb438..dae6e521bb 100644 --- a/esphome/components/bluetooth_proxy/bluetooth_connection.cpp +++ b/esphome/components/bluetooth_proxy/bluetooth_connection.cpp @@ -13,11 +13,180 @@ namespace bluetooth_proxy { static const char *const TAG = "bluetooth_proxy.connection"; +static std::vector get_128bit_uuid_vec(esp_bt_uuid_t uuid_source) { + esp_bt_uuid_t uuid = espbt::ESPBTUUID::from_uuid(uuid_source).as_128bit().get_uuid(); + return std::vector{((uint64_t) uuid.uuid.uuid128[15] << 56) | ((uint64_t) uuid.uuid.uuid128[14] << 48) | + ((uint64_t) uuid.uuid.uuid128[13] << 40) | ((uint64_t) uuid.uuid.uuid128[12] << 32) | + ((uint64_t) uuid.uuid.uuid128[11] << 24) | ((uint64_t) uuid.uuid.uuid128[10] << 16) | + ((uint64_t) uuid.uuid.uuid128[9] << 8) | ((uint64_t) uuid.uuid.uuid128[8]), + ((uint64_t) uuid.uuid.uuid128[7] << 56) | ((uint64_t) uuid.uuid.uuid128[6] << 48) | + ((uint64_t) uuid.uuid.uuid128[5] << 40) | ((uint64_t) uuid.uuid.uuid128[4] << 32) | + ((uint64_t) uuid.uuid.uuid128[3] << 24) | ((uint64_t) uuid.uuid.uuid128[2] << 16) | + ((uint64_t) uuid.uuid.uuid128[1] << 8) | ((uint64_t) uuid.uuid.uuid128[0])}; +} + void BluetoothConnection::dump_config() { ESP_LOGCONFIG(TAG, "BLE Connection:"); BLEClientBase::dump_config(); } +void BluetoothConnection::loop() { + BLEClientBase::loop(); + + // Early return if no active connection or not in service discovery phase + if (this->address_ == 0 || this->send_service_ < 0 || this->send_service_ > this->service_count_) { + return; + } + + // Handle service discovery + this->send_service_for_discovery_(); +} + +void BluetoothConnection::reset_connection_(esp_err_t reason) { + // Send disconnection notification + this->proxy_->send_device_connection(this->address_, false, 0, reason); + + // Important: If we were in the middle of sending services, we do NOT send + // send_gatt_services_done() here. This ensures the client knows that + // the service discovery was interrupted and can retry. The client + // (aioesphomeapi) implements a 30-second timeout (DEFAULT_BLE_TIMEOUT) + // to detect incomplete service discovery rather than relying on us to + // tell them about a partial list. + this->set_address(0); + this->send_service_ = DONE_SENDING_SERVICES; + this->proxy_->send_connections_free(); +} + +void BluetoothConnection::send_service_for_discovery_() { + if (this->send_service_ == this->service_count_) { + this->send_service_ = DONE_SENDING_SERVICES; + this->proxy_->send_gatt_services_done(this->address_); + if (this->connection_type_ == espbt::ConnectionType::V3_WITH_CACHE || + this->connection_type_ == espbt::ConnectionType::V3_WITHOUT_CACHE) { + this->release_services(); + } + return; + } + + // Early return if no API connection + auto *api_conn = this->proxy_->get_api_connection(); + if (api_conn == nullptr) { + return; + } + + // Send next service + esp_gattc_service_elem_t service_result; + uint16_t service_count = 1; + esp_gatt_status_t service_status = esp_ble_gattc_get_service(this->gattc_if_, this->conn_id_, nullptr, + &service_result, &service_count, this->send_service_); + this->send_service_++; + + if (service_status != ESP_GATT_OK) { + ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_service error at offset=%d, status=%d", this->connection_index_, + this->address_str().c_str(), this->send_service_ - 1, service_status); + return; + } + + if (service_count == 0) { + ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_service missing, service_count=%d", this->connection_index_, + this->address_str().c_str(), service_count); + return; + } + + api::BluetoothGATTGetServicesResponse resp; + resp.address = this->address_; + resp.services.reserve(1); // Always one service per response in this implementation + api::BluetoothGATTService service_resp; + service_resp.uuid = get_128bit_uuid_vec(service_result.uuid); + service_resp.handle = service_result.start_handle; + + // Get the number of characteristics directly with one call + uint16_t total_char_count = 0; + esp_gatt_status_t char_count_status = + esp_ble_gattc_get_attr_count(this->gattc_if_, this->conn_id_, ESP_GATT_DB_CHARACTERISTIC, + service_result.start_handle, service_result.end_handle, 0, &total_char_count); + + if (char_count_status == ESP_GATT_OK && total_char_count > 0) { + // Only reserve if we successfully got a count + service_resp.characteristics.reserve(total_char_count); + } else if (char_count_status != ESP_GATT_OK) { + ESP_LOGW(TAG, "[%d] [%s] Error getting characteristic count, status=%d", this->connection_index_, + this->address_str().c_str(), char_count_status); + } + + // Now process characteristics + uint16_t char_offset = 0; + esp_gattc_char_elem_t char_result; + while (true) { // characteristics + uint16_t char_count = 1; + esp_gatt_status_t char_status = + esp_ble_gattc_get_all_char(this->gattc_if_, this->conn_id_, service_result.start_handle, + service_result.end_handle, &char_result, &char_count, char_offset); + if (char_status == ESP_GATT_INVALID_OFFSET || char_status == ESP_GATT_NOT_FOUND) { + break; + } + if (char_status != ESP_GATT_OK) { + ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_all_char error, status=%d", this->connection_index_, + this->address_str().c_str(), char_status); + break; + } + if (char_count == 0) { + break; + } + + api::BluetoothGATTCharacteristic characteristic_resp; + characteristic_resp.uuid = get_128bit_uuid_vec(char_result.uuid); + characteristic_resp.handle = char_result.char_handle; + characteristic_resp.properties = char_result.properties; + char_offset++; + + // Get the number of descriptors directly with one call + uint16_t total_desc_count = 0; + esp_gatt_status_t desc_count_status = + esp_ble_gattc_get_attr_count(this->gattc_if_, this->conn_id_, ESP_GATT_DB_DESCRIPTOR, char_result.char_handle, + service_result.end_handle, 0, &total_desc_count); + + if (desc_count_status == ESP_GATT_OK && total_desc_count > 0) { + // Only reserve if we successfully got a count + characteristic_resp.descriptors.reserve(total_desc_count); + } else if (desc_count_status != ESP_GATT_OK) { + ESP_LOGW(TAG, "[%d] [%s] Error getting descriptor count for char handle %d, status=%d", this->connection_index_, + this->address_str().c_str(), char_result.char_handle, desc_count_status); + } + + // Now process descriptors + uint16_t desc_offset = 0; + esp_gattc_descr_elem_t desc_result; + while (true) { // descriptors + uint16_t desc_count = 1; + esp_gatt_status_t desc_status = esp_ble_gattc_get_all_descr( + this->gattc_if_, this->conn_id_, char_result.char_handle, &desc_result, &desc_count, desc_offset); + if (desc_status == ESP_GATT_INVALID_OFFSET || desc_status == ESP_GATT_NOT_FOUND) { + break; + } + if (desc_status != ESP_GATT_OK) { + ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_all_descr error, status=%d", this->connection_index_, + this->address_str().c_str(), desc_status); + break; + } + if (desc_count == 0) { + break; + } + + api::BluetoothGATTDescriptor descriptor_resp; + descriptor_resp.uuid = get_128bit_uuid_vec(desc_result.uuid); + descriptor_resp.handle = desc_result.handle; + characteristic_resp.descriptors.push_back(std::move(descriptor_resp)); + desc_offset++; + } + service_resp.characteristics.push_back(std::move(characteristic_resp)); + } + resp.services.push_back(std::move(service_resp)); + + // Send the message (we already checked api_conn is not null at the beginning) + api_conn->send_message(resp, api::BluetoothGATTGetServicesResponse::MESSAGE_TYPE); +} + bool BluetoothConnection::gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t gattc_if, esp_ble_gattc_cb_param_t *param) { if (!BLEClientBase::gattc_event_handler(event, gattc_if, param)) @@ -25,22 +194,16 @@ bool BluetoothConnection::gattc_event_handler(esp_gattc_cb_event_t event, esp_ga switch (event) { case ESP_GATTC_DISCONNECT_EVT: { - this->proxy_->send_device_connection(this->address_, false, 0, param->disconnect.reason); - this->set_address(0); - this->proxy_->send_connections_free(); + this->reset_connection_(param->disconnect.reason); break; } case ESP_GATTC_CLOSE_EVT: { - this->proxy_->send_device_connection(this->address_, false, 0, param->close.reason); - this->set_address(0); - this->proxy_->send_connections_free(); + this->reset_connection_(param->close.reason); break; } case ESP_GATTC_OPEN_EVT: { if (param->open.status != ESP_GATT_OK && param->open.status != ESP_GATT_ALREADY_OPEN) { - this->proxy_->send_device_connection(this->address_, false, 0, param->open.status); - this->set_address(0); - this->proxy_->send_connections_free(); + this->reset_connection_(param->open.status); } else if (this->connection_type_ == espbt::ConnectionType::V3_WITH_CACHE) { this->proxy_->send_device_connection(this->address_, true, this->mtu_); this->proxy_->send_connections_free(); diff --git a/esphome/components/bluetooth_proxy/bluetooth_connection.h b/esphome/components/bluetooth_proxy/bluetooth_connection.h index 73c034d93b..2673238fba 100644 --- a/esphome/components/bluetooth_proxy/bluetooth_connection.h +++ b/esphome/components/bluetooth_proxy/bluetooth_connection.h @@ -12,6 +12,7 @@ class BluetoothProxy; class BluetoothConnection : public esp32_ble_client::BLEClientBase { public: void dump_config() override; + void loop() override; bool gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t gattc_if, esp_ble_gattc_cb_param_t *param) override; void gap_event_handler(esp_gap_ble_cb_event_t event, esp_ble_gap_cb_param_t *param) override; @@ -27,6 +28,9 @@ class BluetoothConnection : public esp32_ble_client::BLEClientBase { protected: friend class BluetoothProxy; + void send_service_for_discovery_(); + void reset_connection_(esp_err_t reason); + // Memory optimized layout for 32-bit systems // Group 1: Pointers (4 bytes each, naturally aligned) BluetoothProxy *proxy_; diff --git a/esphome/components/bluetooth_proxy/bluetooth_proxy.cpp b/esphome/components/bluetooth_proxy/bluetooth_proxy.cpp index 7d12842a24..f4b63f3a5d 100644 --- a/esphome/components/bluetooth_proxy/bluetooth_proxy.cpp +++ b/esphome/components/bluetooth_proxy/bluetooth_proxy.cpp @@ -11,19 +11,6 @@ namespace esphome { namespace bluetooth_proxy { static const char *const TAG = "bluetooth_proxy"; -static const int DONE_SENDING_SERVICES = -2; - -std::vector get_128bit_uuid_vec(esp_bt_uuid_t uuid_source) { - esp_bt_uuid_t uuid = espbt::ESPBTUUID::from_uuid(uuid_source).as_128bit().get_uuid(); - return std::vector{((uint64_t) uuid.uuid.uuid128[15] << 56) | ((uint64_t) uuid.uuid.uuid128[14] << 48) | - ((uint64_t) uuid.uuid.uuid128[13] << 40) | ((uint64_t) uuid.uuid.uuid128[12] << 32) | - ((uint64_t) uuid.uuid.uuid128[11] << 24) | ((uint64_t) uuid.uuid.uuid128[10] << 16) | - ((uint64_t) uuid.uuid.uuid128[9] << 8) | ((uint64_t) uuid.uuid.uuid128[8]), - ((uint64_t) uuid.uuid.uuid128[7] << 56) | ((uint64_t) uuid.uuid.uuid128[6] << 48) | - ((uint64_t) uuid.uuid.uuid128[5] << 40) | ((uint64_t) uuid.uuid.uuid128[4] << 32) | - ((uint64_t) uuid.uuid.uuid128[3] << 24) | ((uint64_t) uuid.uuid.uuid128[2] << 16) | - ((uint64_t) uuid.uuid.uuid128[1] << 8) | ((uint64_t) uuid.uuid.uuid128[0])}; -} // Batch size for BLE advertisements to maximize WiFi efficiency // Each advertisement is up to 80 bytes when packaged (including protocol overhead) @@ -213,130 +200,12 @@ void BluetoothProxy::loop() { } // Flush any pending BLE advertisements that have been accumulated but not yet sent - static uint32_t last_flush_time = 0; uint32_t now = App.get_loop_component_start_time(); // Flush accumulated advertisements every 100ms - if (now - last_flush_time >= 100) { + if (now - this->last_advertisement_flush_time_ >= 100) { this->flush_pending_advertisements(); - last_flush_time = now; - } - for (auto *connection : this->connections_) { - if (connection->send_service_ == connection->service_count_) { - connection->send_service_ = DONE_SENDING_SERVICES; - this->send_gatt_services_done(connection->get_address()); - if (connection->connection_type_ == espbt::ConnectionType::V3_WITH_CACHE || - connection->connection_type_ == espbt::ConnectionType::V3_WITHOUT_CACHE) { - connection->release_services(); - } - } else if (connection->send_service_ >= 0) { - esp_gattc_service_elem_t service_result; - uint16_t service_count = 1; - esp_gatt_status_t service_status = - esp_ble_gattc_get_service(connection->get_gattc_if(), connection->get_conn_id(), nullptr, &service_result, - &service_count, connection->send_service_); - connection->send_service_++; - if (service_status != ESP_GATT_OK) { - ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_service error at offset=%d, status=%d", - connection->get_connection_index(), connection->address_str().c_str(), connection->send_service_ - 1, - service_status); - continue; - } - if (service_count == 0) { - ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_service missing, service_count=%d", - connection->get_connection_index(), connection->address_str().c_str(), service_count); - continue; - } - api::BluetoothGATTGetServicesResponse resp; - resp.address = connection->get_address(); - resp.services.reserve(1); // Always one service per response in this implementation - api::BluetoothGATTService service_resp; - service_resp.uuid = get_128bit_uuid_vec(service_result.uuid); - service_resp.handle = service_result.start_handle; - uint16_t char_offset = 0; - esp_gattc_char_elem_t char_result; - // Get the number of characteristics directly with one call - uint16_t total_char_count = 0; - esp_gatt_status_t char_count_status = esp_ble_gattc_get_attr_count( - connection->get_gattc_if(), connection->get_conn_id(), ESP_GATT_DB_CHARACTERISTIC, - service_result.start_handle, service_result.end_handle, 0, &total_char_count); - - if (char_count_status == ESP_GATT_OK && total_char_count > 0) { - // Only reserve if we successfully got a count - service_resp.characteristics.reserve(total_char_count); - } else if (char_count_status != ESP_GATT_OK) { - ESP_LOGW(TAG, "[%d] [%s] Error getting characteristic count, status=%d", connection->get_connection_index(), - connection->address_str().c_str(), char_count_status); - } - - // Now process characteristics - while (true) { // characteristics - uint16_t char_count = 1; - esp_gatt_status_t char_status = esp_ble_gattc_get_all_char( - connection->get_gattc_if(), connection->get_conn_id(), service_result.start_handle, - service_result.end_handle, &char_result, &char_count, char_offset); - if (char_status == ESP_GATT_INVALID_OFFSET || char_status == ESP_GATT_NOT_FOUND) { - break; - } - if (char_status != ESP_GATT_OK) { - ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_all_char error, status=%d", connection->get_connection_index(), - connection->address_str().c_str(), char_status); - break; - } - if (char_count == 0) { - break; - } - api::BluetoothGATTCharacteristic characteristic_resp; - characteristic_resp.uuid = get_128bit_uuid_vec(char_result.uuid); - characteristic_resp.handle = char_result.char_handle; - characteristic_resp.properties = char_result.properties; - char_offset++; - - // Get the number of descriptors directly with one call - uint16_t total_desc_count = 0; - esp_gatt_status_t desc_count_status = - esp_ble_gattc_get_attr_count(connection->get_gattc_if(), connection->get_conn_id(), ESP_GATT_DB_DESCRIPTOR, - char_result.char_handle, service_result.end_handle, 0, &total_desc_count); - - if (desc_count_status == ESP_GATT_OK && total_desc_count > 0) { - // Only reserve if we successfully got a count - characteristic_resp.descriptors.reserve(total_desc_count); - } else if (desc_count_status != ESP_GATT_OK) { - ESP_LOGW(TAG, "[%d] [%s] Error getting descriptor count for char handle %d, status=%d", - connection->get_connection_index(), connection->address_str().c_str(), char_result.char_handle, - desc_count_status); - } - - // Now process descriptors - uint16_t desc_offset = 0; - esp_gattc_descr_elem_t desc_result; - while (true) { // descriptors - uint16_t desc_count = 1; - esp_gatt_status_t desc_status = - esp_ble_gattc_get_all_descr(connection->get_gattc_if(), connection->get_conn_id(), - char_result.char_handle, &desc_result, &desc_count, desc_offset); - if (desc_status == ESP_GATT_INVALID_OFFSET || desc_status == ESP_GATT_NOT_FOUND) { - break; - } - if (desc_status != ESP_GATT_OK) { - ESP_LOGE(TAG, "[%d] [%s] esp_ble_gattc_get_all_descr error, status=%d", connection->get_connection_index(), - connection->address_str().c_str(), desc_status); - break; - } - if (desc_count == 0) { - break; - } - api::BluetoothGATTDescriptor descriptor_resp; - descriptor_resp.uuid = get_128bit_uuid_vec(desc_result.uuid); - descriptor_resp.handle = desc_result.handle; - characteristic_resp.descriptors.push_back(std::move(descriptor_resp)); - desc_offset++; - } - service_resp.characteristics.push_back(std::move(characteristic_resp)); - } - resp.services.push_back(std::move(service_resp)); - this->api_connection_->send_message(resp, api::BluetoothGATTGetServicesResponse::MESSAGE_TYPE); - } + this->last_advertisement_flush_time_ = now; } } diff --git a/esphome/components/bluetooth_proxy/bluetooth_proxy.h b/esphome/components/bluetooth_proxy/bluetooth_proxy.h index 52f1d0f88a..9d84a9dbf2 100644 --- a/esphome/components/bluetooth_proxy/bluetooth_proxy.h +++ b/esphome/components/bluetooth_proxy/bluetooth_proxy.h @@ -22,6 +22,7 @@ namespace esphome { namespace bluetooth_proxy { static const esp_err_t ESP_GATT_NOT_CONNECTED = -1; +static const int DONE_SENDING_SERVICES = -2; using namespace esp32_ble_client; @@ -149,7 +150,10 @@ class BluetoothProxy : public esp32_ble_tracker::ESPBTDeviceListener, public Com std::vector advertisement_pool_; std::unique_ptr response_; - // Group 3: 1-byte types grouped together + // Group 3: 4-byte types + uint32_t last_advertisement_flush_time_{0}; + + // Group 4: 1-byte types grouped together bool active_; uint8_t advertisement_count_{0}; // 2 bytes used, 2 bytes padding From 534a1cf2e72e0efc392f571477ff1c1817c6ea43 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 20 Jul 2025 12:17:38 -1000 Subject: [PATCH 3/4] [esp32_ble_tracker] Batch BLE advertisement processing to reduce overhead (#9699) --- .../esp32_ble_tracker/esp32_ble_tracker.cpp | 49 +++++++++++-------- 1 file changed, 28 insertions(+), 21 deletions(-) diff --git a/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp b/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp index 44577afbbd..96003073d7 100644 --- a/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp +++ b/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp @@ -128,46 +128,53 @@ void ESP32BLETracker::loop() { uint8_t write_idx = this->ring_write_index_.load(std::memory_order_acquire); while (read_idx != write_idx) { - // Process one result at a time directly from ring buffer - BLEScanResult &scan_result = this->scan_ring_buffer_[read_idx]; + // Calculate how many contiguous results we can process in one batch + // If write > read: process all results from read to write + // If write <= read (wraparound): process from read to end of buffer first + size_t batch_size = (write_idx > read_idx) ? (write_idx - read_idx) : (SCAN_RESULT_BUFFER_SIZE - read_idx); + // Process the batch for raw advertisements if (this->raw_advertisements_) { for (auto *listener : this->listeners_) { - listener->parse_devices(&scan_result, 1); + listener->parse_devices(&this->scan_ring_buffer_[read_idx], batch_size); } for (auto *client : this->clients_) { - client->parse_devices(&scan_result, 1); + client->parse_devices(&this->scan_ring_buffer_[read_idx], batch_size); } } + // Process individual results for parsed advertisements if (this->parse_advertisements_) { #ifdef USE_ESP32_BLE_DEVICE - ESPBTDevice device; - device.parse_scan_rst(scan_result); + for (size_t i = 0; i < batch_size; i++) { + BLEScanResult &scan_result = this->scan_ring_buffer_[read_idx + i]; + ESPBTDevice device; + device.parse_scan_rst(scan_result); - bool found = false; - for (auto *listener : this->listeners_) { - if (listener->parse_device(device)) - found = true; - } + bool found = false; + for (auto *listener : this->listeners_) { + if (listener->parse_device(device)) + found = true; + } - for (auto *client : this->clients_) { - if (client->parse_device(device)) { - found = true; - if (!connecting && client->state() == ClientState::DISCOVERED) { - promote_to_connecting = true; + for (auto *client : this->clients_) { + if (client->parse_device(device)) { + found = true; + if (!connecting && client->state() == ClientState::DISCOVERED) { + promote_to_connecting = true; + } } } - } - if (!found && !this->scan_continuous_) { - this->print_bt_device_info(device); + if (!found && !this->scan_continuous_) { + this->print_bt_device_info(device); + } } #endif // USE_ESP32_BLE_DEVICE } - // Move to next entry in ring buffer - read_idx = (read_idx + 1) % SCAN_RESULT_BUFFER_SIZE; + // Update read index for entire batch + read_idx = (read_idx + batch_size) % 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); From e474a33abd984872273f1fd2942a34f54a2389d0 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 20 Jul 2025 12:20:35 -1000 Subject: [PATCH 4/4] [api] Memory optimizations for API frame helper buffering (#9724) --- esphome/components/api/api_frame_helper.cpp | 78 ++++++++++----------- esphome/components/api/api_frame_helper.h | 13 ++-- 2 files changed, 45 insertions(+), 46 deletions(-) diff --git a/esphome/components/api/api_frame_helper.cpp b/esphome/components/api/api_frame_helper.cpp index afd64e8981..2e7956cb74 100644 --- a/esphome/components/api/api_frame_helper.cpp +++ b/esphome/components/api/api_frame_helper.cpp @@ -77,32 +77,45 @@ APIError APIFrameHelper::loop() { } // Helper method to buffer data from IOVs -void APIFrameHelper::buffer_data_from_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len) { +void APIFrameHelper::buffer_data_from_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len, + uint16_t offset) { SendBuffer buffer; - buffer.data.reserve(total_write_len); + buffer.size = total_write_len - offset; + buffer.data = std::make_unique(buffer.size); + + uint16_t to_skip = offset; + uint16_t write_pos = 0; + for (int i = 0; i < iovcnt; i++) { - const uint8_t *data = reinterpret_cast(iov[i].iov_base); - buffer.data.insert(buffer.data.end(), data, data + iov[i].iov_len); + if (to_skip >= iov[i].iov_len) { + // Skip this entire segment + to_skip -= static_cast(iov[i].iov_len); + } else { + // Include this segment (partially or fully) + const uint8_t *src = reinterpret_cast(iov[i].iov_base) + to_skip; + uint16_t len = static_cast(iov[i].iov_len) - to_skip; + std::memcpy(buffer.data.get() + write_pos, src, len); + write_pos += len; + to_skip = 0; + } } this->tx_buf_.push_back(std::move(buffer)); } // This method writes data to socket or buffers it -APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt) { +APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt, uint16_t total_write_len) { // Returns APIError::OK if successful (or would block, but data has been buffered) // Returns APIError::SOCKET_WRITE_FAILED if socket write failed, and sets state to FAILED if (iovcnt == 0) return APIError::OK; // Nothing to do, success - uint16_t total_write_len = 0; - for (int i = 0; i < iovcnt; i++) { #ifdef HELPER_LOG_PACKETS + for (int i = 0; i < iovcnt; i++) { ESP_LOGVV(TAG, "Sending raw: %s", format_hex_pretty(reinterpret_cast(iov[i].iov_base), iov[i].iov_len).c_str()); -#endif - total_write_len += static_cast(iov[i].iov_len); } +#endif // Try to send any existing buffered data first if there is any if (!this->tx_buf_.empty()) { @@ -115,7 +128,7 @@ APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt) { // If there is still data in the buffer, we can't send, buffer // the new data and return if (!this->tx_buf_.empty()) { - this->buffer_data_from_iov_(iov, iovcnt, total_write_len); + this->buffer_data_from_iov_(iov, iovcnt, total_write_len, 0); return APIError::OK; // Success, data buffered } } @@ -126,7 +139,7 @@ APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt) { if (sent == -1) { if (errno == EWOULDBLOCK || errno == EAGAIN) { // Socket would block, buffer the data - this->buffer_data_from_iov_(iov, iovcnt, total_write_len); + this->buffer_data_from_iov_(iov, iovcnt, total_write_len, 0); return APIError::OK; // Success, data buffered } // Socket error @@ -135,26 +148,7 @@ APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt) { return APIError::SOCKET_WRITE_FAILED; // Socket write failed } else if (static_cast(sent) < total_write_len) { // Partially sent, buffer the remaining data - SendBuffer buffer; - uint16_t to_consume = static_cast(sent); - uint16_t remaining = total_write_len - static_cast(sent); - - buffer.data.reserve(remaining); - - for (int i = 0; i < iovcnt; i++) { - if (to_consume >= iov[i].iov_len) { - // This segment was fully sent - to_consume -= static_cast(iov[i].iov_len); - } else { - // This segment was partially sent or not sent at all - const uint8_t *data = reinterpret_cast(iov[i].iov_base) + to_consume; - uint16_t len = static_cast(iov[i].iov_len) - to_consume; - buffer.data.insert(buffer.data.end(), data, data + len); - to_consume = 0; - } - } - - this->tx_buf_.push_back(std::move(buffer)); + this->buffer_data_from_iov_(iov, iovcnt, total_write_len, static_cast(sent)); } return APIError::OK; // Success, all data sent or buffered @@ -639,6 +633,7 @@ APIError APINoiseFrameHelper::write_protobuf_packets(ProtoWriteBuffer buffer, st this->reusable_iovs_.clear(); this->reusable_iovs_.reserve(packets.size()); + uint16_t total_write_len = 0; // We need to encrypt each packet in place for (const auto &packet : packets) { @@ -678,12 +673,13 @@ APIError APINoiseFrameHelper::write_protobuf_packets(ProtoWriteBuffer buffer, st buf_start[2] = static_cast(mbuf.size); // Add iovec for this encrypted packet - this->reusable_iovs_.push_back( - {buf_start, static_cast(3 + mbuf.size)}); // indicator + size + encrypted data + size_t packet_len = static_cast(3 + mbuf.size); // indicator + size + encrypted data + this->reusable_iovs_.push_back({buf_start, packet_len}); + total_write_len += packet_len; } // Send all encrypted packets in one writev call - return this->write_raw_(this->reusable_iovs_.data(), this->reusable_iovs_.size()); + return this->write_raw_(this->reusable_iovs_.data(), this->reusable_iovs_.size(), total_write_len); } APIError APINoiseFrameHelper::write_frame_(const uint8_t *data, uint16_t len) { @@ -696,12 +692,12 @@ APIError APINoiseFrameHelper::write_frame_(const uint8_t *data, uint16_t len) { iov[0].iov_base = header; iov[0].iov_len = 3; if (len == 0) { - return this->write_raw_(iov, 1); + return this->write_raw_(iov, 1, 3); // Just header } iov[1].iov_base = const_cast(data); iov[1].iov_len = len; - return this->write_raw_(iov, 2); + return this->write_raw_(iov, 2, 3 + len); // Header + data } /** Initiate the data structures for the handshake. @@ -990,7 +986,7 @@ APIError APIPlaintextFrameHelper::read_packet(ReadPacketBuffer *buffer) { "Bad indicator byte"; iov[0].iov_base = (void *) msg; iov[0].iov_len = 19; - this->write_raw_(iov, 1); + this->write_raw_(iov, 1, 19); } return aerr; } @@ -1020,6 +1016,7 @@ APIError APIPlaintextFrameHelper::write_protobuf_packets(ProtoWriteBuffer buffer this->reusable_iovs_.clear(); this->reusable_iovs_.reserve(packets.size()); + uint16_t total_write_len = 0; for (const auto &packet : packets) { // Calculate varint sizes for header layout @@ -1064,12 +1061,13 @@ APIError APIPlaintextFrameHelper::write_protobuf_packets(ProtoWriteBuffer buffer .encode_to_buffer_unchecked(buf_start + header_offset + 1 + size_varint_len, type_varint_len); // Add iovec for this packet (header + payload) - this->reusable_iovs_.push_back( - {buf_start + header_offset, static_cast(total_header_len + packet.payload_size)}); + size_t packet_len = static_cast(total_header_len + packet.payload_size); + this->reusable_iovs_.push_back({buf_start + header_offset, packet_len}); + total_write_len += packet_len; } // Send all packets in one writev call - return write_raw_(this->reusable_iovs_.data(), this->reusable_iovs_.size()); + return write_raw_(this->reusable_iovs_.data(), this->reusable_iovs_.size(), total_write_len); } #endif // USE_API_PLAINTEXT diff --git a/esphome/components/api/api_frame_helper.h b/esphome/components/api/api_frame_helper.h index 4bcc4acd61..b5b25700a8 100644 --- a/esphome/components/api/api_frame_helper.h +++ b/esphome/components/api/api_frame_helper.h @@ -116,22 +116,23 @@ class APIFrameHelper { // Buffer containing data to be sent struct SendBuffer { - std::vector data; - uint16_t offset{0}; // Current offset within the buffer (uint16_t to reduce memory usage) + std::unique_ptr data; + uint16_t size{0}; // Total size of the buffer + uint16_t offset{0}; // Current offset within the buffer // Using uint16_t reduces memory usage since ESPHome API messages are limited to UINT16_MAX (65535) bytes - uint16_t remaining() const { return static_cast(data.size()) - offset; } - const uint8_t *current_data() const { return data.data() + offset; } + uint16_t remaining() const { return size - offset; } + const uint8_t *current_data() const { return data.get() + offset; } }; // Common implementation for writing raw data to socket - APIError write_raw_(const struct iovec *iov, int iovcnt); + APIError write_raw_(const struct iovec *iov, int iovcnt, uint16_t total_write_len); // Try to send data from the tx buffer APIError try_send_tx_buf_(); // Helper method to buffer data from IOVs - void buffer_data_from_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len); + void buffer_data_from_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len, uint16_t offset); template APIError write_raw_(const struct iovec *iov, int iovcnt, socket::Socket *socket, std::vector &tx_buf, const std::string &info, StateEnum &state, StateEnum failed_state);