mirror of
https://github.com/esphome/esphome.git
synced 2026-09-25 14:00:25 +00:00
[scheduler][core] Lock-free fast-path on ESPHOME_THREAD_MULTI_NO_ATOMICS via __atomic builtins (#15947)
This commit is contained in:
+10
-10
@@ -235,11 +235,11 @@ void HOT Scheduler::set_timer_common_(Component *component, SchedulerItem::Type
|
||||
}
|
||||
target->push_back(item);
|
||||
if (target == &this->to_add_) {
|
||||
this->to_add_count_increment_();
|
||||
this->to_add_count_increment_locked_();
|
||||
}
|
||||
#ifndef ESPHOME_THREAD_SINGLE
|
||||
else {
|
||||
this->defer_count_increment_();
|
||||
this->defer_count_increment_locked_();
|
||||
}
|
||||
#endif
|
||||
}
|
||||
@@ -452,7 +452,7 @@ void Scheduler::full_cleanup_removed_items_() {
|
||||
this->items_.erase(this->items_.begin() + write, this->items_.end());
|
||||
// Rebuild the heap structure since items are no longer in heap order
|
||||
std::make_heap(this->items_.begin(), this->items_.end(), SchedulerItem::cmp);
|
||||
this->to_remove_clear_();
|
||||
this->to_remove_clear_locked_();
|
||||
}
|
||||
|
||||
#ifndef ESPHOME_THREAD_SINGLE
|
||||
@@ -501,7 +501,7 @@ void HOT Scheduler::process_defer_queue_slow_path_(uint32_t &now) {
|
||||
|
||||
this->lock_.lock();
|
||||
// Reset counter and snapshot queue end under lock
|
||||
this->defer_count_clear_();
|
||||
this->defer_count_clear_locked_();
|
||||
size_t defer_queue_end = this->defer_queue_.size();
|
||||
if (this->defer_queue_front_ >= defer_queue_end) {
|
||||
this->lock_.unlock();
|
||||
@@ -621,7 +621,7 @@ uint32_t HOT Scheduler::call(uint32_t now) {
|
||||
LockGuard guard{this->lock_};
|
||||
if (is_item_removed_locked_(item)) {
|
||||
this->recycle_item_main_loop_(this->pop_raw_locked_());
|
||||
this->to_remove_decrement_();
|
||||
this->to_remove_decrement_locked_();
|
||||
continue;
|
||||
}
|
||||
}
|
||||
@@ -630,7 +630,7 @@ uint32_t HOT Scheduler::call(uint32_t now) {
|
||||
if (is_item_removed_(item)) {
|
||||
LockGuard guard{this->lock_};
|
||||
this->recycle_item_main_loop_(this->pop_raw_locked_());
|
||||
this->to_remove_decrement_();
|
||||
this->to_remove_decrement_locked_();
|
||||
continue;
|
||||
}
|
||||
#endif
|
||||
@@ -658,7 +658,7 @@ uint32_t HOT Scheduler::call(uint32_t now) {
|
||||
|
||||
if (this->is_item_removed_locked_(executed_item)) {
|
||||
// We were removed/cancelled in the function call, recycle and continue
|
||||
this->to_remove_decrement_();
|
||||
this->to_remove_decrement_locked_();
|
||||
this->recycle_item_main_loop_(executed_item);
|
||||
continue;
|
||||
}
|
||||
@@ -721,7 +721,7 @@ void HOT Scheduler::process_to_add_slow_path_() {
|
||||
std::push_heap(this->items_.begin(), this->items_.end(), SchedulerItem::cmp);
|
||||
}
|
||||
this->to_add_.clear();
|
||||
this->to_add_count_clear_();
|
||||
this->to_add_count_clear_locked_();
|
||||
}
|
||||
bool HOT Scheduler::cleanup_slow_path_() {
|
||||
// We must hold the lock for the entire cleanup operation because:
|
||||
@@ -737,7 +737,7 @@ bool HOT Scheduler::cleanup_slow_path_() {
|
||||
SchedulerItem *item = this->items_[0];
|
||||
if (!this->is_item_removed_locked_(item))
|
||||
break;
|
||||
this->to_remove_decrement_();
|
||||
this->to_remove_decrement_locked_();
|
||||
this->recycle_item_main_loop_(this->pop_raw_locked_());
|
||||
}
|
||||
return !this->items_.empty();
|
||||
@@ -825,7 +825,7 @@ bool HOT Scheduler::cancel_item_locked_(Component *component, NameType name_type
|
||||
size_t heap_cancelled = this->mark_matching_items_removed_locked_(this->items_, component, name_type, static_name,
|
||||
hash_or_id, type, match_retry, find_first);
|
||||
total_cancelled += heap_cancelled;
|
||||
this->to_remove_add_(heap_cancelled);
|
||||
this->to_remove_add_locked_(heap_cancelled);
|
||||
if (find_first && total_cancelled > 0)
|
||||
return true;
|
||||
}
|
||||
|
||||
+57
-43
@@ -524,11 +524,13 @@ class Scheduler {
|
||||
std::vector<SchedulerItem *> to_add_;
|
||||
|
||||
#ifndef ESPHOME_THREAD_SINGLE
|
||||
// Fast-path counter for process_to_add() to skip taking the lock when there is
|
||||
// nothing to add. Uses std::atomic on platforms that support it, plain uint32_t
|
||||
// otherwise. On non-atomic platforms, callers must hold the scheduler lock when
|
||||
// mutating this counter. Not needed on single-threaded platforms where we can
|
||||
// check to_add_.empty() directly.
|
||||
// Fast-path counter for process_to_add() to skip taking the lock when there
|
||||
// is nothing to add. std::atomic on ATOMICS; plain uint32_t on NO_ATOMICS
|
||||
// (BK72xx — ARMv5TE single-core, lacks LDREX/STREX so std::atomic RMW would
|
||||
// require libatomic). Reads use __atomic_load_n(__ATOMIC_RELAXED) on
|
||||
// NO_ATOMICS — compiles to a plain LDR (aligned 32-bit load is naturally
|
||||
// atomic on ARMv5TE) but expresses the concurrent-access intent in the C++
|
||||
// memory model. Writes live behind *_locked_ helpers and must hold lock_.
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
std::atomic<uint32_t> to_add_count_{0};
|
||||
#else
|
||||
@@ -536,40 +538,41 @@ class Scheduler {
|
||||
#endif
|
||||
#endif /* ESPHOME_THREAD_SINGLE */
|
||||
|
||||
// Fast-path helper for process_to_add() to decide if it can try the lock-free path.
|
||||
// - On ESPHOME_THREAD_SINGLE: direct container check is safe (no concurrent writers).
|
||||
// - On ESPHOME_THREAD_MULTI_ATOMICS: performs a lock-free check via to_add_count_.
|
||||
// - On ESPHOME_THREAD_MULTI_NO_ATOMICS: always returns false to force the caller
|
||||
// down the locked path; this is NOT a lock-free emptiness check on that platform.
|
||||
// Fast-path helper for process_to_add() to decide if it can skip the lock.
|
||||
bool to_add_empty_() const {
|
||||
#ifdef ESPHOME_THREAD_SINGLE
|
||||
return this->to_add_.empty();
|
||||
#elif defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
return this->to_add_count_.load(std::memory_order_relaxed) == 0;
|
||||
#else
|
||||
return false;
|
||||
return __atomic_load_n(&this->to_add_count_, __ATOMIC_RELAXED) == 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
// Increment to_add_count_ (no-op on single-threaded platforms)
|
||||
void to_add_count_increment_() {
|
||||
#ifdef ESPHOME_THREAD_SINGLE
|
||||
// Increment to_add_count_ (no-op on single-threaded platforms).
|
||||
// On NO_ATOMICS the caller must hold lock_; both load and store go through
|
||||
// __atomic_*_n with __ATOMIC_RELAXED to keep every access to the counter
|
||||
// explicitly atomic in the C++ memory model (same ARMv5TE codegen as
|
||||
// plain LDR+STR).
|
||||
void to_add_count_increment_locked_() {
|
||||
#if defined(ESPHOME_THREAD_SINGLE)
|
||||
// No counter needed — to_add_empty_() checks the vector directly
|
||||
#elif defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
this->to_add_count_.fetch_add(1, std::memory_order_relaxed);
|
||||
#else
|
||||
this->to_add_count_++;
|
||||
uint32_t v = __atomic_load_n(&this->to_add_count_, __ATOMIC_RELAXED);
|
||||
__atomic_store_n(&this->to_add_count_, v + 1, __ATOMIC_RELAXED);
|
||||
#endif
|
||||
}
|
||||
|
||||
// Reset to_add_count_ (no-op on single-threaded platforms)
|
||||
void to_add_count_clear_() {
|
||||
#ifdef ESPHOME_THREAD_SINGLE
|
||||
void to_add_count_clear_locked_() {
|
||||
#if defined(ESPHOME_THREAD_SINGLE)
|
||||
// No counter needed — to_add_empty_() checks the vector directly
|
||||
#elif defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
this->to_add_count_.store(0, std::memory_order_relaxed);
|
||||
#else
|
||||
this->to_add_count_ = 0;
|
||||
__atomic_store_n(&this->to_add_count_, 0, __ATOMIC_RELAXED);
|
||||
#endif
|
||||
}
|
||||
|
||||
@@ -580,7 +583,8 @@ class Scheduler {
|
||||
std::vector<SchedulerItem *> defer_queue_; // FIFO queue for defer() calls
|
||||
size_t defer_queue_front_{0}; // Index of first valid item in defer_queue_ (tracks consumed items)
|
||||
|
||||
// Fast-path counter for process_defer_queue_() to skip lock when nothing to process.
|
||||
// Fast-path counter for process_defer_queue_() to skip lock when nothing to
|
||||
// process. See to_add_count_ above for the NO_ATOMICS rationale.
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
std::atomic<uint32_t> defer_count_{0};
|
||||
#else
|
||||
@@ -589,35 +593,35 @@ class Scheduler {
|
||||
|
||||
bool defer_empty_() const {
|
||||
// defer_queue_ only exists on multi-threaded platforms, so no ESPHOME_THREAD_SINGLE path
|
||||
// ESPHOME_THREAD_MULTI_NO_ATOMICS: always take the lock
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
return this->defer_count_.load(std::memory_order_relaxed) == 0;
|
||||
#else
|
||||
return false;
|
||||
return __atomic_load_n(&this->defer_count_, __ATOMIC_RELAXED) == 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
void defer_count_increment_() {
|
||||
void defer_count_increment_locked_() {
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
this->defer_count_.fetch_add(1, std::memory_order_relaxed);
|
||||
#else
|
||||
this->defer_count_++;
|
||||
uint32_t v = __atomic_load_n(&this->defer_count_, __ATOMIC_RELAXED);
|
||||
__atomic_store_n(&this->defer_count_, v + 1, __ATOMIC_RELAXED);
|
||||
#endif
|
||||
}
|
||||
|
||||
void defer_count_clear_() {
|
||||
void defer_count_clear_locked_() {
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
this->defer_count_.store(0, std::memory_order_relaxed);
|
||||
#else
|
||||
this->defer_count_ = 0;
|
||||
__atomic_store_n(&this->defer_count_, 0, __ATOMIC_RELAXED);
|
||||
#endif
|
||||
}
|
||||
|
||||
#endif /* ESPHOME_THREAD_SINGLE */
|
||||
|
||||
// Counter for items marked for removal. Incremented cross-thread in cancel_item_locked_().
|
||||
// On ESPHOME_THREAD_MULTI_ATOMICS this is read without a lock in the cleanup_() fast path;
|
||||
// on ESPHOME_THREAD_MULTI_NO_ATOMICS the fast path is disabled so cleanup_() always takes the lock.
|
||||
// Counter for items marked for removal. Incremented cross-thread in
|
||||
// cancel_item_locked_(). See to_add_count_ above for the NO_ATOMICS
|
||||
// rationale.
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
std::atomic<uint32_t> to_remove_{0};
|
||||
#else
|
||||
@@ -626,44 +630,54 @@ class Scheduler {
|
||||
|
||||
// Lock-free check if there are items to remove (for fast-path in cleanup_)
|
||||
bool to_remove_empty_() const {
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
#if defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
return this->to_remove_.load(std::memory_order_relaxed) == 0;
|
||||
#elif defined(ESPHOME_THREAD_SINGLE)
|
||||
return this->to_remove_ == 0;
|
||||
#elif defined(ESPHOME_THREAD_MULTI_NO_ATOMICS)
|
||||
return __atomic_load_n(&this->to_remove_, __ATOMIC_RELAXED) == 0;
|
||||
#else
|
||||
return false; // Always take the lock path
|
||||
return this->to_remove_ == 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
void to_remove_add_(uint32_t count) {
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
void to_remove_add_locked_(uint32_t count) {
|
||||
#if defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
this->to_remove_.fetch_add(count, std::memory_order_relaxed);
|
||||
#elif defined(ESPHOME_THREAD_MULTI_NO_ATOMICS)
|
||||
uint32_t v = __atomic_load_n(&this->to_remove_, __ATOMIC_RELAXED);
|
||||
__atomic_store_n(&this->to_remove_, v + count, __ATOMIC_RELAXED);
|
||||
#else
|
||||
this->to_remove_ += count;
|
||||
this->to_remove_ += count;
|
||||
#endif
|
||||
}
|
||||
|
||||
void to_remove_decrement_() {
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
void to_remove_decrement_locked_() {
|
||||
#if defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
this->to_remove_.fetch_sub(1, std::memory_order_relaxed);
|
||||
#elif defined(ESPHOME_THREAD_MULTI_NO_ATOMICS)
|
||||
uint32_t v = __atomic_load_n(&this->to_remove_, __ATOMIC_RELAXED);
|
||||
__atomic_store_n(&this->to_remove_, v - 1, __ATOMIC_RELAXED);
|
||||
#else
|
||||
this->to_remove_--;
|
||||
this->to_remove_--;
|
||||
#endif
|
||||
}
|
||||
|
||||
void to_remove_clear_() {
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
void to_remove_clear_locked_() {
|
||||
#if defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
this->to_remove_.store(0, std::memory_order_relaxed);
|
||||
#elif defined(ESPHOME_THREAD_MULTI_NO_ATOMICS)
|
||||
__atomic_store_n(&this->to_remove_, 0, __ATOMIC_RELAXED);
|
||||
#else
|
||||
this->to_remove_ = 0;
|
||||
this->to_remove_ = 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
uint32_t to_remove_count_() const {
|
||||
#ifdef ESPHOME_THREAD_MULTI_ATOMICS
|
||||
#if defined(ESPHOME_THREAD_MULTI_ATOMICS)
|
||||
return this->to_remove_.load(std::memory_order_relaxed);
|
||||
#elif defined(ESPHOME_THREAD_MULTI_NO_ATOMICS)
|
||||
return __atomic_load_n(&this->to_remove_, __ATOMIC_RELAXED);
|
||||
#else
|
||||
return this->to_remove_;
|
||||
return this->to_remove_;
|
||||
#endif
|
||||
}
|
||||
|
||||
|
||||
@@ -74,8 +74,8 @@ uint64_t Millis64Impl::compute(uint32_t now) {
|
||||
// 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 = millis_major;
|
||||
uint32_t last = last_millis;
|
||||
uint16_t major = __atomic_load_n(&millis_major, __ATOMIC_RELAXED);
|
||||
uint32_t last = __atomic_load_n(&last_millis, __ATOMIC_RELAXED);
|
||||
|
||||
// Define a safe window around the rollover point (10 seconds)
|
||||
// This covers any reasonable scheduler delays or thread preemption
|
||||
@@ -87,19 +87,26 @@ uint64_t Millis64Impl::compute(uint32_t now) {
|
||||
if (near_rollover || (now < last && (last - now) > HALF_MAX_UINT32)) {
|
||||
// Near rollover or detected a rollover - need lock for safety
|
||||
LockGuard guard{lock};
|
||||
// Re-read with lock held
|
||||
last = last_millis;
|
||||
// Re-read both values with lock held. last_millis can be updated
|
||||
// unlocked from the forward-progression branch below, so use an atomic
|
||||
// load. millis_major can only be updated under this lock, but another
|
||||
// thread may have completed a rollover between our unlocked loads above
|
||||
// and the lock acquisition — reload or we'd return a stale high word.
|
||||
last = __atomic_load_n(&last_millis, __ATOMIC_RELAXED);
|
||||
major = __atomic_load_n(&millis_major, __ATOMIC_RELAXED);
|
||||
|
||||
if (now < last && (last - now) > HALF_MAX_UINT32) {
|
||||
// True rollover detected (happens every ~49.7 days)
|
||||
millis_major++;
|
||||
// True rollover detected (happens every ~49.7 days).
|
||||
// Use the already-loaded `major` local; avoids a second read of the
|
||||
// global (equivalent under the held lock).
|
||||
major++;
|
||||
__atomic_store_n(&millis_major, major, __ATOMIC_RELAXED);
|
||||
#ifdef ESPHOME_DEBUG_SCHEDULER
|
||||
ESP_LOGD(TAG, "Detected true 32-bit rollover at %" PRIu32 "ms (was %" PRIu32 ")", now, last);
|
||||
#endif /* ESPHOME_DEBUG_SCHEDULER */
|
||||
}
|
||||
// Update last_millis while holding lock
|
||||
last_millis = now;
|
||||
__atomic_store_n(&last_millis, now, __ATOMIC_RELAXED);
|
||||
} else if (now > last) {
|
||||
// Normal case: Not near rollover and time moved forward
|
||||
// Update without lock. While this may cause minor races (microseconds of
|
||||
@@ -107,7 +114,7 @@ uint64_t Millis64Impl::compute(uint32_t now) {
|
||||
// 1. The scheduler operates at millisecond resolution, not microsecond
|
||||
// 2. We've already prevented the critical rollover race condition
|
||||
// 3. Any backwards movement is orders of magnitude smaller than scheduler delays
|
||||
last_millis = now;
|
||||
__atomic_store_n(&last_millis, now, __ATOMIC_RELAXED);
|
||||
}
|
||||
// If now <= last and we're not near rollover, don't update
|
||||
// This minimizes backwards time movement
|
||||
|
||||
Reference in New Issue
Block a user