mirror of
				https://github.com/esphome/esphome.git
				synced 2025-10-31 07:03:55 +00:00 
			
		
		
		
	Add heap scheduler tests
This commit is contained in:
		| @@ -0,0 +1,21 @@ | ||||
| import esphome.codegen as cg | ||||
| import esphome.config_validation as cv | ||||
| from esphome.const import CONF_ID | ||||
|  | ||||
| scheduler_heap_stress_component_ns = cg.esphome_ns.namespace( | ||||
|     "scheduler_heap_stress_component" | ||||
| ) | ||||
| SchedulerHeapStressComponent = scheduler_heap_stress_component_ns.class_( | ||||
|     "SchedulerHeapStressComponent", cg.Component | ||||
| ) | ||||
|  | ||||
| CONFIG_SCHEMA = cv.Schema( | ||||
|     { | ||||
|         cv.GenerateID(): cv.declare_id(SchedulerHeapStressComponent), | ||||
|     } | ||||
| ).extend(cv.COMPONENT_SCHEMA) | ||||
|  | ||||
|  | ||||
| async def to_code(config): | ||||
|     var = cg.new_Pvariable(config[CONF_ID]) | ||||
|     await cg.register_component(var, config) | ||||
| @@ -0,0 +1,104 @@ | ||||
| #include "heap_scheduler_stress_component.h" | ||||
| #include "esphome/core/log.h" | ||||
| #include <thread> | ||||
| #include <atomic> | ||||
| #include <vector> | ||||
| #include <chrono> | ||||
| #include <random> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_heap_stress_component { | ||||
|  | ||||
| static const char *const TAG = "scheduler_heap_stress"; | ||||
|  | ||||
| void SchedulerHeapStressComponent::setup() { ESP_LOGCONFIG(TAG, "SchedulerHeapStressComponent setup"); } | ||||
|  | ||||
| void SchedulerHeapStressComponent::run_multi_thread_test() { | ||||
|   // Use member variables instead of static to avoid issues | ||||
|   this->total_callbacks_ = 0; | ||||
|   this->executed_callbacks_ = 0; | ||||
|   static constexpr int NUM_THREADS = 10; | ||||
|   static constexpr int CALLBACKS_PER_THREAD = 100; | ||||
|  | ||||
|   ESP_LOGI(TAG, "Starting heap scheduler stress test - multi-threaded concurrent set_timeout/set_interval"); | ||||
|  | ||||
|   // Ensure we're starting clean | ||||
|   ESP_LOGI(TAG, "Initial counters: total=%d, executed=%d", this->total_callbacks_.load(), | ||||
|            this->executed_callbacks_.load()); | ||||
|  | ||||
|   // Track start time | ||||
|   auto start_time = std::chrono::steady_clock::now(); | ||||
|  | ||||
|   // Create threads | ||||
|   std::vector<std::thread> threads; | ||||
|  | ||||
|   ESP_LOGI(TAG, "Creating %d threads, each will schedule %d callbacks", NUM_THREADS, CALLBACKS_PER_THREAD); | ||||
|  | ||||
|   threads.reserve(NUM_THREADS); | ||||
|   for (int i = 0; i < NUM_THREADS; i++) { | ||||
|     threads.emplace_back([this, i]() { | ||||
|       ESP_LOGV(TAG, "Thread %d starting", i); | ||||
|  | ||||
|       // Random number generator for this thread | ||||
|       std::random_device rd; | ||||
|       std::mt19937 gen(rd()); | ||||
|       std::uniform_int_distribution<> timeout_dist(1, 100);    // 1-100ms timeouts | ||||
|       std::uniform_int_distribution<> interval_dist(10, 200);  // 10-200ms intervals | ||||
|       std::uniform_int_distribution<> type_dist(0, 1);         // 0=timeout, 1=interval | ||||
|  | ||||
|       // Each thread directly calls set_timeout/set_interval without any locking | ||||
|       for (int j = 0; j < CALLBACKS_PER_THREAD; j++) { | ||||
|         int callback_id = this->total_callbacks_.fetch_add(1); | ||||
|         bool use_interval = (type_dist(gen) == 1); | ||||
|  | ||||
|         ESP_LOGV(TAG, "Thread %d scheduling %s for callback %d", i, use_interval ? "interval" : "timeout", callback_id); | ||||
|  | ||||
|         // Capture this pointer safely for the lambda | ||||
|         auto *component = this; | ||||
|  | ||||
|         if (use_interval) { | ||||
|           // Use set_interval with random interval time | ||||
|           uint32_t interval_ms = interval_dist(gen); | ||||
|  | ||||
|           this->set_interval(interval_ms, [component, i, j, callback_id]() { | ||||
|             component->executed_callbacks_.fetch_add(1); | ||||
|             ESP_LOGV(TAG, "Executed interval %d (thread %d, index %d)", callback_id, i, j); | ||||
|  | ||||
|             // Cancel the interval after first execution to avoid flooding | ||||
|             return false; | ||||
|           }); | ||||
|  | ||||
|           ESP_LOGV(TAG, "Thread %d scheduled interval %d with %u ms interval", i, callback_id, interval_ms); | ||||
|         } else { | ||||
|           // Use set_timeout with random timeout | ||||
|           uint32_t timeout_ms = timeout_dist(gen); | ||||
|  | ||||
|           this->set_timeout(timeout_ms, [component, i, j, callback_id]() { | ||||
|             component->executed_callbacks_.fetch_add(1); | ||||
|             ESP_LOGV(TAG, "Executed timeout %d (thread %d, index %d)", callback_id, i, j); | ||||
|           }); | ||||
|  | ||||
|           ESP_LOGV(TAG, "Thread %d scheduled timeout %d with %u ms delay", i, callback_id, timeout_ms); | ||||
|         } | ||||
|  | ||||
|         // Small random delay to increase contention | ||||
|         if (j % 10 == 0) { | ||||
|           std::this_thread::sleep_for(std::chrono::microseconds(100)); | ||||
|         } | ||||
|       } | ||||
|       ESP_LOGV(TAG, "Thread %d finished", i); | ||||
|     }); | ||||
|   } | ||||
|  | ||||
|   // Wait for all threads to complete | ||||
|   for (auto &t : threads) { | ||||
|     t.join(); | ||||
|   } | ||||
|  | ||||
|   auto end_time = std::chrono::steady_clock::now(); | ||||
|   auto thread_time = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count(); | ||||
|   ESP_LOGI(TAG, "All threads finished in %lldms. Created %d callbacks", thread_time, this->total_callbacks_.load()); | ||||
| } | ||||
|  | ||||
| }  // namespace scheduler_heap_stress_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,22 @@ | ||||
| #pragma once | ||||
|  | ||||
| #include "esphome/core/component.h" | ||||
| #include <atomic> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_heap_stress_component { | ||||
|  | ||||
| class SchedulerHeapStressComponent : public Component { | ||||
|  public: | ||||
|   void setup() override; | ||||
|   float get_setup_priority() const override { return setup_priority::LATE; } | ||||
|  | ||||
|   void run_multi_thread_test(); | ||||
|  | ||||
|  private: | ||||
|   std::atomic<int> total_callbacks_{0}; | ||||
|   std::atomic<int> executed_callbacks_{0}; | ||||
| }; | ||||
|  | ||||
| }  // namespace scheduler_heap_stress_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,21 @@ | ||||
| import esphome.codegen as cg | ||||
| import esphome.config_validation as cv | ||||
| from esphome.const import CONF_ID | ||||
|  | ||||
| scheduler_rapid_cancellation_component_ns = cg.esphome_ns.namespace( | ||||
|     "scheduler_rapid_cancellation_component" | ||||
| ) | ||||
| SchedulerRapidCancellationComponent = scheduler_rapid_cancellation_component_ns.class_( | ||||
|     "SchedulerRapidCancellationComponent", cg.Component | ||||
| ) | ||||
|  | ||||
| CONFIG_SCHEMA = cv.Schema( | ||||
|     { | ||||
|         cv.GenerateID(): cv.declare_id(SchedulerRapidCancellationComponent), | ||||
|     } | ||||
| ).extend(cv.COMPONENT_SCHEMA) | ||||
|  | ||||
|  | ||||
| async def to_code(config): | ||||
|     var = cg.new_Pvariable(config[CONF_ID]) | ||||
|     await cg.register_component(var, config) | ||||
| @@ -0,0 +1,77 @@ | ||||
| #include "rapid_cancellation_component.h" | ||||
| #include "esphome/core/log.h" | ||||
| #include <thread> | ||||
| #include <vector> | ||||
| #include <chrono> | ||||
| #include <random> | ||||
| #include <sstream> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_rapid_cancellation_component { | ||||
|  | ||||
| static const char *const TAG = "scheduler_rapid_cancellation"; | ||||
|  | ||||
| void SchedulerRapidCancellationComponent::setup() { ESP_LOGCONFIG(TAG, "SchedulerRapidCancellationComponent setup"); } | ||||
|  | ||||
| void SchedulerRapidCancellationComponent::run_rapid_cancellation_test() { | ||||
|   ESP_LOGI(TAG, "Starting rapid cancellation test - multiple threads racing on same timeout names"); | ||||
|  | ||||
|   // Reset counters | ||||
|   this->total_scheduled_ = 0; | ||||
|   this->total_executed_ = 0; | ||||
|  | ||||
|   static constexpr int NUM_THREADS = 4;              // Number of threads to create | ||||
|   static constexpr int NUM_NAMES = 10;               // Only 10 unique names | ||||
|   static constexpr int OPERATIONS_PER_THREAD = 100;  // Each thread does 100 operations | ||||
|  | ||||
|   // Create threads that will all fight over the same timeout names | ||||
|   std::vector<std::thread> threads; | ||||
|   threads.reserve(NUM_THREADS); | ||||
|  | ||||
|   for (int thread_id = 0; thread_id < NUM_THREADS; thread_id++) { | ||||
|     threads.emplace_back([this, thread_id]() { | ||||
|       for (int i = 0; i < OPERATIONS_PER_THREAD; i++) { | ||||
|         // Use modulo to ensure multiple threads use the same names | ||||
|         int name_index = i % NUM_NAMES; | ||||
|         std::stringstream ss; | ||||
|         ss << "shared_timeout_" << name_index; | ||||
|         std::string name = ss.str(); | ||||
|  | ||||
|         // All threads schedule timeouts - this will implicitly cancel existing ones | ||||
|         this->set_timeout(name, 100, [this, name]() { | ||||
|           this->total_executed_.fetch_add(1); | ||||
|           ESP_LOGI(TAG, "Executed callback '%s'", name.c_str()); | ||||
|         }); | ||||
|         this->total_scheduled_.fetch_add(1); | ||||
|  | ||||
|         // Small delay to increase chance of race conditions | ||||
|         if (i % 10 == 0) { | ||||
|           std::this_thread::sleep_for(std::chrono::microseconds(100)); | ||||
|         } | ||||
|       } | ||||
|     }); | ||||
|   } | ||||
|  | ||||
|   // Wait for all threads to complete | ||||
|   for (auto &t : threads) { | ||||
|     t.join(); | ||||
|   } | ||||
|  | ||||
|   ESP_LOGI(TAG, "All threads completed. Scheduled: %d", this->total_scheduled_.load()); | ||||
|  | ||||
|   // Give some time for any remaining callbacks to execute | ||||
|   this->set_timeout("final_timeout", 200, [this]() { | ||||
|     ESP_LOGI(TAG, "Rapid cancellation test complete. Final stats:"); | ||||
|     ESP_LOGI(TAG, "  Total scheduled: %d", this->total_scheduled_.load()); | ||||
|     ESP_LOGI(TAG, "  Total executed: %d", this->total_executed_.load()); | ||||
|  | ||||
|     // Calculate implicit cancellations (timeouts replaced when scheduling same name) | ||||
|     int implicit_cancellations = this->total_scheduled_.load() - this->total_executed_.load(); | ||||
|     ESP_LOGI(TAG, "  Implicit cancellations (replaced): %d", implicit_cancellations); | ||||
|     ESP_LOGI(TAG, "  Total accounted: %d (executed + implicit cancellations)", | ||||
|              this->total_executed_.load() + implicit_cancellations); | ||||
|   }); | ||||
| } | ||||
|  | ||||
| }  // namespace scheduler_rapid_cancellation_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,22 @@ | ||||
| #pragma once | ||||
|  | ||||
| #include "esphome/core/component.h" | ||||
| #include <atomic> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_rapid_cancellation_component { | ||||
|  | ||||
| class SchedulerRapidCancellationComponent : public Component { | ||||
|  public: | ||||
|   void setup() override; | ||||
|   float get_setup_priority() const override { return setup_priority::LATE; } | ||||
|  | ||||
|   void run_rapid_cancellation_test(); | ||||
|  | ||||
|  private: | ||||
|   std::atomic<int> total_scheduled_{0}; | ||||
|   std::atomic<int> total_executed_{0}; | ||||
| }; | ||||
|  | ||||
| }  // namespace scheduler_rapid_cancellation_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,21 @@ | ||||
| import esphome.codegen as cg | ||||
| import esphome.config_validation as cv | ||||
| from esphome.const import CONF_ID | ||||
|  | ||||
| scheduler_recursive_timeout_component_ns = cg.esphome_ns.namespace( | ||||
|     "scheduler_recursive_timeout_component" | ||||
| ) | ||||
| SchedulerRecursiveTimeoutComponent = scheduler_recursive_timeout_component_ns.class_( | ||||
|     "SchedulerRecursiveTimeoutComponent", cg.Component | ||||
| ) | ||||
|  | ||||
| CONFIG_SCHEMA = cv.Schema( | ||||
|     { | ||||
|         cv.GenerateID(): cv.declare_id(SchedulerRecursiveTimeoutComponent), | ||||
|     } | ||||
| ).extend(cv.COMPONENT_SCHEMA) | ||||
|  | ||||
|  | ||||
| async def to_code(config): | ||||
|     var = cg.new_Pvariable(config[CONF_ID]) | ||||
|     await cg.register_component(var, config) | ||||
| @@ -0,0 +1,40 @@ | ||||
| #include "recursive_timeout_component.h" | ||||
| #include "esphome/core/log.h" | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_recursive_timeout_component { | ||||
|  | ||||
| static const char *const TAG = "scheduler_recursive_timeout"; | ||||
|  | ||||
| void SchedulerRecursiveTimeoutComponent::setup() { ESP_LOGCONFIG(TAG, "SchedulerRecursiveTimeoutComponent setup"); } | ||||
|  | ||||
| void SchedulerRecursiveTimeoutComponent::run_recursive_timeout_test() { | ||||
|   ESP_LOGI(TAG, "Starting recursive timeout test - scheduling timeout from within timeout"); | ||||
|  | ||||
|   // Reset state | ||||
|   this->nested_level_ = 0; | ||||
|  | ||||
|   // Schedule the initial timeout with 1ms delay | ||||
|   this->set_timeout(1, [this]() { | ||||
|     ESP_LOGI(TAG, "Executing initial timeout"); | ||||
|     this->nested_level_ = 1; | ||||
|  | ||||
|     // From within this timeout, schedule another timeout with 1ms delay | ||||
|     this->set_timeout(1, [this]() { | ||||
|       ESP_LOGI(TAG, "Executing nested timeout 1"); | ||||
|       this->nested_level_ = 2; | ||||
|  | ||||
|       // From within this nested timeout, schedule yet another timeout with 1ms delay | ||||
|       this->set_timeout(1, [this]() { | ||||
|         ESP_LOGI(TAG, "Executing nested timeout 2"); | ||||
|         this->nested_level_ = 3; | ||||
|  | ||||
|         // Test complete | ||||
|         ESP_LOGI(TAG, "Recursive timeout test complete - all %d levels executed", this->nested_level_); | ||||
|       }); | ||||
|     }); | ||||
|   }); | ||||
| } | ||||
|  | ||||
| }  // namespace scheduler_recursive_timeout_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,20 @@ | ||||
| #pragma once | ||||
|  | ||||
| #include "esphome/core/component.h" | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_recursive_timeout_component { | ||||
|  | ||||
| class SchedulerRecursiveTimeoutComponent : public Component { | ||||
|  public: | ||||
|   void setup() override; | ||||
|   float get_setup_priority() const override { return setup_priority::LATE; } | ||||
|  | ||||
|   void run_recursive_timeout_test(); | ||||
|  | ||||
|  private: | ||||
|   int nested_level_{0}; | ||||
| }; | ||||
|  | ||||
| }  // namespace scheduler_recursive_timeout_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,23 @@ | ||||
| import esphome.codegen as cg | ||||
| import esphome.config_validation as cv | ||||
| from esphome.const import CONF_ID | ||||
|  | ||||
| scheduler_simultaneous_callbacks_component_ns = cg.esphome_ns.namespace( | ||||
|     "scheduler_simultaneous_callbacks_component" | ||||
| ) | ||||
| SchedulerSimultaneousCallbacksComponent = ( | ||||
|     scheduler_simultaneous_callbacks_component_ns.class_( | ||||
|         "SchedulerSimultaneousCallbacksComponent", cg.Component | ||||
|     ) | ||||
| ) | ||||
|  | ||||
| CONFIG_SCHEMA = cv.Schema( | ||||
|     { | ||||
|         cv.GenerateID(): cv.declare_id(SchedulerSimultaneousCallbacksComponent), | ||||
|     } | ||||
| ).extend(cv.COMPONENT_SCHEMA) | ||||
|  | ||||
|  | ||||
| async def to_code(config): | ||||
|     var = cg.new_Pvariable(config[CONF_ID]) | ||||
|     await cg.register_component(var, config) | ||||
| @@ -0,0 +1,109 @@ | ||||
| #include "simultaneous_callbacks_component.h" | ||||
| #include "esphome/core/log.h" | ||||
| #include <thread> | ||||
| #include <vector> | ||||
| #include <chrono> | ||||
| #include <sstream> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_simultaneous_callbacks_component { | ||||
|  | ||||
| static const char *const TAG = "scheduler_simultaneous_callbacks"; | ||||
|  | ||||
| void SchedulerSimultaneousCallbacksComponent::setup() { | ||||
|   ESP_LOGCONFIG(TAG, "SchedulerSimultaneousCallbacksComponent setup"); | ||||
| } | ||||
|  | ||||
| void SchedulerSimultaneousCallbacksComponent::run_simultaneous_callbacks_test() { | ||||
|   ESP_LOGI(TAG, "Starting simultaneous callbacks test - 10 threads scheduling 100 callbacks each for 1ms from now"); | ||||
|  | ||||
|   // Reset counters | ||||
|   this->total_scheduled_ = 0; | ||||
|   this->total_executed_ = 0; | ||||
|   this->callbacks_at_once_ = 0; | ||||
|   this->max_concurrent_ = 0; | ||||
|  | ||||
|   static constexpr int NUM_THREADS = 10; | ||||
|   static constexpr int CALLBACKS_PER_THREAD = 100; | ||||
|   static constexpr uint32_t DELAY_MS = 1;  // All callbacks scheduled for 1ms from now | ||||
|  | ||||
|   // Create threads for concurrent scheduling | ||||
|   std::vector<std::thread> threads; | ||||
|   threads.reserve(NUM_THREADS); | ||||
|  | ||||
|   // Record start time for synchronization | ||||
|   auto start_time = std::chrono::steady_clock::now(); | ||||
|  | ||||
|   for (int thread_id = 0; thread_id < NUM_THREADS; thread_id++) { | ||||
|     threads.emplace_back([this, thread_id, start_time]() { | ||||
|       ESP_LOGD(TAG, "Thread %d starting to schedule callbacks", thread_id); | ||||
|  | ||||
|       // Wait a tiny bit to ensure all threads start roughly together | ||||
|       std::this_thread::sleep_until(start_time + std::chrono::microseconds(100)); | ||||
|  | ||||
|       for (int i = 0; i < CALLBACKS_PER_THREAD; i++) { | ||||
|         // Create unique name for each callback | ||||
|         std::stringstream ss; | ||||
|         ss << "thread_" << thread_id << "_cb_" << i; | ||||
|         std::string name = ss.str(); | ||||
|  | ||||
|         // Schedule callback for exactly DELAY_MS from now | ||||
|         this->set_timeout(name, DELAY_MS, [this, thread_id, i, name]() { | ||||
|           // Increment concurrent counter atomically | ||||
|           int current = this->callbacks_at_once_.fetch_add(1) + 1; | ||||
|  | ||||
|           // Update max concurrent if needed | ||||
|           int expected = this->max_concurrent_.load(); | ||||
|           while (current > expected && !this->max_concurrent_.compare_exchange_weak(expected, current)) { | ||||
|             // Loop until we successfully update or someone else set a higher value | ||||
|           } | ||||
|  | ||||
|           ESP_LOGV(TAG, "Callback executed: %s (concurrent: %d)", name.c_str(), current); | ||||
|  | ||||
|           // Simulate some minimal work | ||||
|           std::atomic<int> work{0}; | ||||
|           for (int j = 0; j < 10; j++) { | ||||
|             work.fetch_add(j); | ||||
|           } | ||||
|  | ||||
|           // Increment executed counter | ||||
|           this->total_executed_.fetch_add(1); | ||||
|  | ||||
|           // Decrement concurrent counter | ||||
|           this->callbacks_at_once_.fetch_sub(1); | ||||
|         }); | ||||
|  | ||||
|         this->total_scheduled_.fetch_add(1); | ||||
|         ESP_LOGV(TAG, "Scheduled callback %s", name.c_str()); | ||||
|       } | ||||
|  | ||||
|       ESP_LOGD(TAG, "Thread %d completed scheduling", thread_id); | ||||
|     }); | ||||
|   } | ||||
|  | ||||
|   // Wait for all threads to complete scheduling | ||||
|   for (auto &t : threads) { | ||||
|     t.join(); | ||||
|   } | ||||
|  | ||||
|   ESP_LOGI(TAG, "All threads completed scheduling. Total scheduled: %d", this->total_scheduled_.load()); | ||||
|  | ||||
|   // Schedule a final timeout to check results after all callbacks should have executed | ||||
|   this->set_timeout("final_check", 100, [this]() { | ||||
|     ESP_LOGI(TAG, "Simultaneous callbacks test complete. Final executed count: %d", this->total_executed_.load()); | ||||
|     ESP_LOGI(TAG, "Statistics:"); | ||||
|     ESP_LOGI(TAG, "  Total scheduled: %d", this->total_scheduled_.load()); | ||||
|     ESP_LOGI(TAG, "  Total executed: %d", this->total_executed_.load()); | ||||
|     ESP_LOGI(TAG, "  Max concurrent callbacks: %d", this->max_concurrent_.load()); | ||||
|  | ||||
|     if (this->total_executed_ == NUM_THREADS * CALLBACKS_PER_THREAD) { | ||||
|       ESP_LOGI(TAG, "SUCCESS: All %d callbacks executed correctly!", this->total_executed_.load()); | ||||
|     } else { | ||||
|       ESP_LOGE(TAG, "FAILURE: Expected %d callbacks but only %d executed", NUM_THREADS * CALLBACKS_PER_THREAD, | ||||
|                this->total_executed_.load()); | ||||
|     } | ||||
|   }); | ||||
| } | ||||
|  | ||||
| }  // namespace scheduler_simultaneous_callbacks_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,24 @@ | ||||
| #pragma once | ||||
|  | ||||
| #include "esphome/core/component.h" | ||||
| #include <atomic> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_simultaneous_callbacks_component { | ||||
|  | ||||
| class SchedulerSimultaneousCallbacksComponent : public Component { | ||||
|  public: | ||||
|   void setup() override; | ||||
|   float get_setup_priority() const override { return setup_priority::LATE; } | ||||
|  | ||||
|   void run_simultaneous_callbacks_test(); | ||||
|  | ||||
|  private: | ||||
|   std::atomic<int> total_scheduled_{0}; | ||||
|   std::atomic<int> total_executed_{0}; | ||||
|   std::atomic<int> callbacks_at_once_{0}; | ||||
|   std::atomic<int> max_concurrent_{0}; | ||||
| }; | ||||
|  | ||||
| }  // namespace scheduler_simultaneous_callbacks_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,21 @@ | ||||
| import esphome.codegen as cg | ||||
| import esphome.config_validation as cv | ||||
| from esphome.const import CONF_ID | ||||
|  | ||||
| scheduler_string_lifetime_component_ns = cg.esphome_ns.namespace( | ||||
|     "scheduler_string_lifetime_component" | ||||
| ) | ||||
| SchedulerStringLifetimeComponent = scheduler_string_lifetime_component_ns.class_( | ||||
|     "SchedulerStringLifetimeComponent", cg.Component | ||||
| ) | ||||
|  | ||||
| CONFIG_SCHEMA = cv.Schema( | ||||
|     { | ||||
|         cv.GenerateID(): cv.declare_id(SchedulerStringLifetimeComponent), | ||||
|     } | ||||
| ).extend(cv.COMPONENT_SCHEMA) | ||||
|  | ||||
|  | ||||
| async def to_code(config): | ||||
|     var = cg.new_Pvariable(config[CONF_ID]) | ||||
|     await cg.register_component(var, config) | ||||
| @@ -0,0 +1,233 @@ | ||||
| #include "string_lifetime_component.h" | ||||
| #include "esphome/core/log.h" | ||||
| #include <memory> | ||||
| #include <thread> | ||||
| #include <chrono> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_string_lifetime_component { | ||||
|  | ||||
| static const char *const TAG = "scheduler_string_lifetime"; | ||||
|  | ||||
| void SchedulerStringLifetimeComponent::setup() { ESP_LOGCONFIG(TAG, "SchedulerStringLifetimeComponent setup"); } | ||||
|  | ||||
| void SchedulerStringLifetimeComponent::run_string_lifetime_test() { | ||||
|   ESP_LOGI(TAG, "Starting string lifetime tests"); | ||||
|  | ||||
|   this->tests_passed_ = 0; | ||||
|   this->tests_failed_ = 0; | ||||
|  | ||||
|   // Run each test | ||||
|   test_temporary_string_lifetime(); | ||||
|   test_scope_exit_string(); | ||||
|   test_vector_reallocation(); | ||||
|   test_string_move_semantics(); | ||||
|   test_lambda_capture_lifetime(); | ||||
|  | ||||
|   // Schedule final check | ||||
|   this->set_timeout("final_check", 200, [this]() { | ||||
|     ESP_LOGI(TAG, "String lifetime tests complete"); | ||||
|     ESP_LOGI(TAG, "Tests passed: %d", this->tests_passed_); | ||||
|     ESP_LOGI(TAG, "Tests failed: %d", this->tests_failed_); | ||||
|  | ||||
|     if (this->tests_failed_ == 0) { | ||||
|       ESP_LOGI(TAG, "SUCCESS: All string lifetime tests passed!"); | ||||
|     } else { | ||||
|       ESP_LOGE(TAG, "FAILURE: %d string lifetime tests failed!", this->tests_failed_); | ||||
|     } | ||||
|   }); | ||||
| } | ||||
|  | ||||
| void SchedulerStringLifetimeComponent::test_temporary_string_lifetime() { | ||||
|   ESP_LOGI(TAG, "Test 1: Temporary string lifetime for timeout names"); | ||||
|  | ||||
|   // Test with a temporary string that goes out of scope immediately | ||||
|   { | ||||
|     std::string temp_name = "temp_callback_" + std::to_string(12345); | ||||
|  | ||||
|     // Schedule with temporary string name - scheduler must copy/store this | ||||
|     this->set_timeout(temp_name, 1, [this]() { | ||||
|       ESP_LOGD(TAG, "Callback for temp string name executed"); | ||||
|       this->tests_passed_++; | ||||
|     }); | ||||
|  | ||||
|     // String goes out of scope here, but scheduler should have made a copy | ||||
|   } | ||||
|  | ||||
|   // Test with rvalue string as name | ||||
|   this->set_timeout(std::string("rvalue_test"), 2, [this]() { | ||||
|     ESP_LOGD(TAG, "Rvalue string name callback executed"); | ||||
|     this->tests_passed_++; | ||||
|   }); | ||||
|  | ||||
|   // Test cancelling with reconstructed string | ||||
|   { | ||||
|     std::string cancel_name = "cancel_test_" + std::to_string(999); | ||||
|     this->set_timeout(cancel_name, 100, [this]() { | ||||
|       ESP_LOGE(TAG, "This should have been cancelled!"); | ||||
|       this->tests_failed_++; | ||||
|     }); | ||||
|   }  // cancel_name goes out of scope | ||||
|  | ||||
|   // Reconstruct the same string to cancel | ||||
|   std::string cancel_name_2 = "cancel_test_" + std::to_string(999); | ||||
|   bool cancelled = this->cancel_timeout(cancel_name_2); | ||||
|   if (cancelled) { | ||||
|     ESP_LOGD(TAG, "Successfully cancelled with reconstructed string"); | ||||
|     this->tests_passed_++; | ||||
|   } else { | ||||
|     ESP_LOGE(TAG, "Failed to cancel with reconstructed string"); | ||||
|     this->tests_failed_++; | ||||
|   } | ||||
| } | ||||
|  | ||||
| void SchedulerStringLifetimeComponent::test_scope_exit_string() { | ||||
|   ESP_LOGI(TAG, "Test 2: Scope exit string names"); | ||||
|  | ||||
|   // Create string names in a limited scope | ||||
|   { | ||||
|     std::string scoped_name = "scoped_timeout_" + std::to_string(555); | ||||
|  | ||||
|     // Schedule with scoped string name | ||||
|     this->set_timeout(scoped_name, 3, [this]() { | ||||
|       ESP_LOGD(TAG, "Scoped name callback executed"); | ||||
|       this->tests_passed_++; | ||||
|     }); | ||||
|  | ||||
|     // scoped_name goes out of scope here | ||||
|   } | ||||
|  | ||||
|   // Test with dynamically allocated string name | ||||
|   { | ||||
|     auto *dynamic_name = new std::string("dynamic_timeout_" + std::to_string(777)); | ||||
|  | ||||
|     this->set_timeout(*dynamic_name, 4, [this, dynamic_name]() { | ||||
|       ESP_LOGD(TAG, "Dynamic string name callback executed"); | ||||
|       this->tests_passed_++; | ||||
|       delete dynamic_name;  // Clean up in callback | ||||
|     }); | ||||
|  | ||||
|     // Pointer goes out of scope but string object remains until callback | ||||
|   } | ||||
|  | ||||
|   // Test multiple timeouts with same dynamically created name | ||||
|   for (int i = 0; i < 3; i++) { | ||||
|     std::string loop_name = "loop_timeout_" + std::to_string(i); | ||||
|     this->set_timeout(loop_name, 5 + i * 1, [this, i]() { | ||||
|       ESP_LOGD(TAG, "Loop timeout %d executed", i); | ||||
|       this->tests_passed_++; | ||||
|     }); | ||||
|     // loop_name destroyed and recreated each iteration | ||||
|   } | ||||
| } | ||||
|  | ||||
| void SchedulerStringLifetimeComponent::test_vector_reallocation() { | ||||
|   ESP_LOGI(TAG, "Test 3: Vector reallocation stress on timeout names"); | ||||
|  | ||||
|   // Create a vector that will reallocate | ||||
|   std::vector<std::string> names; | ||||
|   names.reserve(2);  // Small initial capacity to force reallocation | ||||
|  | ||||
|   // Schedule callbacks with string names from vector | ||||
|   for (int i = 0; i < 10; i++) { | ||||
|     names.push_back("vector_cb_" + std::to_string(i)); | ||||
|     // Use the string from vector as timeout name | ||||
|     this->set_timeout(names.back(), 8 + i * 1, [this, i]() { | ||||
|       ESP_LOGV(TAG, "Vector name callback %d executed", i); | ||||
|       this->tests_passed_++; | ||||
|     }); | ||||
|   } | ||||
|  | ||||
|   // Force reallocation by adding more elements | ||||
|   // This will move all strings to new memory locations | ||||
|   for (int i = 10; i < 50; i++) { | ||||
|     names.push_back("realloc_trigger_" + std::to_string(i)); | ||||
|   } | ||||
|  | ||||
|   // Add more timeouts after reallocation to ensure old names still work | ||||
|   for (int i = 50; i < 55; i++) { | ||||
|     names.push_back("post_realloc_" + std::to_string(i)); | ||||
|     this->set_timeout(names.back(), 20 + (i - 50), [this]() { | ||||
|       ESP_LOGV(TAG, "Post-reallocation callback executed"); | ||||
|       this->tests_passed_++; | ||||
|     }); | ||||
|   } | ||||
|  | ||||
|   // Clear the vector while timeouts are still pending | ||||
|   names.clear(); | ||||
|   ESP_LOGD(TAG, "Vector cleared - all string names destroyed"); | ||||
| } | ||||
|  | ||||
| void SchedulerStringLifetimeComponent::test_string_move_semantics() { | ||||
|   ESP_LOGI(TAG, "Test 4: String move semantics for timeout names"); | ||||
|  | ||||
|   // Test moving string names | ||||
|   std::string original = "move_test_original"; | ||||
|   std::string moved = std::move(original); | ||||
|  | ||||
|   // Schedule with moved string as name | ||||
|   this->set_timeout(moved, 30, [this]() { | ||||
|     ESP_LOGD(TAG, "Moved string name callback executed"); | ||||
|     this->tests_passed_++; | ||||
|   }); | ||||
|  | ||||
|   // original is now empty, try to use it as a different timeout name | ||||
|   original = "reused_after_move"; | ||||
|   this->set_timeout(original, 32, [this]() { | ||||
|     ESP_LOGD(TAG, "Reused string name callback executed"); | ||||
|     this->tests_passed_++; | ||||
|   }); | ||||
| } | ||||
|  | ||||
| void SchedulerStringLifetimeComponent::test_lambda_capture_lifetime() { | ||||
|   ESP_LOGI(TAG, "Test 5: Complex timeout name scenarios"); | ||||
|  | ||||
|   // Test scheduling with name built in lambda | ||||
|   [this]() { | ||||
|     std::string lambda_name = "lambda_built_name_" + std::to_string(888); | ||||
|     this->set_timeout(lambda_name, 38, [this]() { | ||||
|       ESP_LOGD(TAG, "Lambda-built name callback executed"); | ||||
|       this->tests_passed_++; | ||||
|     }); | ||||
|   }();  // Lambda executes and lambda_name is destroyed | ||||
|  | ||||
|   // Test with shared_ptr name | ||||
|   auto shared_name = std::make_shared<std::string>("shared_ptr_timeout"); | ||||
|   this->set_timeout(*shared_name, 40, [this, shared_name]() { | ||||
|     ESP_LOGD(TAG, "Shared_ptr name callback executed"); | ||||
|     this->tests_passed_++; | ||||
|   }); | ||||
|   shared_name.reset();  // Release the shared_ptr | ||||
|  | ||||
|   // Test overwriting timeout with same name | ||||
|   std::string overwrite_name = "overwrite_test"; | ||||
|   this->set_timeout(overwrite_name, 1000, [this]() { | ||||
|     ESP_LOGE(TAG, "This should have been overwritten!"); | ||||
|     this->tests_failed_++; | ||||
|   }); | ||||
|  | ||||
|   // Overwrite with shorter timeout | ||||
|   this->set_timeout(overwrite_name, 42, [this]() { | ||||
|     ESP_LOGD(TAG, "Overwritten timeout executed"); | ||||
|     this->tests_passed_++; | ||||
|   }); | ||||
|  | ||||
|   // Test very long string name | ||||
|   std::string long_name; | ||||
|   for (int i = 0; i < 100; i++) { | ||||
|     long_name += "very_long_timeout_name_segment_" + std::to_string(i) + "_"; | ||||
|   } | ||||
|   this->set_timeout(long_name, 44, [this]() { | ||||
|     ESP_LOGD(TAG, "Very long name timeout executed"); | ||||
|     this->tests_passed_++; | ||||
|   }); | ||||
|  | ||||
|   // Test empty string as name | ||||
|   this->set_timeout("", 46, [this]() { | ||||
|     ESP_LOGD(TAG, "Empty string name timeout executed"); | ||||
|     this->tests_passed_++; | ||||
|   }); | ||||
| } | ||||
|  | ||||
| }  // namespace scheduler_string_lifetime_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,29 @@ | ||||
| #pragma once | ||||
|  | ||||
| #include "esphome/core/component.h" | ||||
| #include <vector> | ||||
| #include <string> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_string_lifetime_component { | ||||
|  | ||||
| class SchedulerStringLifetimeComponent : public Component { | ||||
|  public: | ||||
|   void setup() override; | ||||
|   float get_setup_priority() const override { return setup_priority::LATE; } | ||||
|  | ||||
|   void run_string_lifetime_test(); | ||||
|  | ||||
|  private: | ||||
|   void test_temporary_string_lifetime(); | ||||
|   void test_scope_exit_string(); | ||||
|   void test_vector_reallocation(); | ||||
|   void test_string_move_semantics(); | ||||
|   void test_lambda_capture_lifetime(); | ||||
|  | ||||
|   int tests_passed_{0}; | ||||
|   int tests_failed_{0}; | ||||
| }; | ||||
|  | ||||
| }  // namespace scheduler_string_lifetime_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,21 @@ | ||||
| import esphome.codegen as cg | ||||
| import esphome.config_validation as cv | ||||
| from esphome.const import CONF_ID | ||||
|  | ||||
| scheduler_string_name_stress_component_ns = cg.esphome_ns.namespace( | ||||
|     "scheduler_string_name_stress_component" | ||||
| ) | ||||
| SchedulerStringNameStressComponent = scheduler_string_name_stress_component_ns.class_( | ||||
|     "SchedulerStringNameStressComponent", cg.Component | ||||
| ) | ||||
|  | ||||
| CONFIG_SCHEMA = cv.Schema( | ||||
|     { | ||||
|         cv.GenerateID(): cv.declare_id(SchedulerStringNameStressComponent), | ||||
|     } | ||||
| ).extend(cv.COMPONENT_SCHEMA) | ||||
|  | ||||
|  | ||||
| async def to_code(config): | ||||
|     var = cg.new_Pvariable(config[CONF_ID]) | ||||
|     await cg.register_component(var, config) | ||||
| @@ -0,0 +1,110 @@ | ||||
| #include "string_name_stress_component.h" | ||||
| #include "esphome/core/log.h" | ||||
| #include <thread> | ||||
| #include <atomic> | ||||
| #include <vector> | ||||
| #include <chrono> | ||||
| #include <string> | ||||
| #include <sstream> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_string_name_stress_component { | ||||
|  | ||||
| static const char *const TAG = "scheduler_string_name_stress"; | ||||
|  | ||||
| void SchedulerStringNameStressComponent::setup() { ESP_LOGCONFIG(TAG, "SchedulerStringNameStressComponent setup"); } | ||||
|  | ||||
| void SchedulerStringNameStressComponent::run_string_name_stress_test() { | ||||
|   // Use member variables to reset state | ||||
|   this->total_callbacks_ = 0; | ||||
|   this->executed_callbacks_ = 0; | ||||
|   static constexpr int NUM_THREADS = 10; | ||||
|   static constexpr int CALLBACKS_PER_THREAD = 100; | ||||
|  | ||||
|   ESP_LOGI(TAG, "Starting string name stress test - multi-threaded set_timeout with std::string names"); | ||||
|   ESP_LOGI(TAG, "This test specifically uses dynamic string names to test memory management"); | ||||
|  | ||||
|   // Track start time | ||||
|   auto start_time = std::chrono::steady_clock::now(); | ||||
|  | ||||
|   // Create threads | ||||
|   std::vector<std::thread> threads; | ||||
|  | ||||
|   ESP_LOGI(TAG, "Creating %d threads, each will schedule %d callbacks with dynamic names", NUM_THREADS, | ||||
|            CALLBACKS_PER_THREAD); | ||||
|  | ||||
|   threads.reserve(NUM_THREADS); | ||||
|   for (int i = 0; i < NUM_THREADS; i++) { | ||||
|     threads.emplace_back([this, i]() { | ||||
|       ESP_LOGV(TAG, "Thread %d starting", i); | ||||
|  | ||||
|       // Each thread schedules callbacks with dynamically created string names | ||||
|       for (int j = 0; j < CALLBACKS_PER_THREAD; j++) { | ||||
|         int callback_id = this->total_callbacks_.fetch_add(1); | ||||
|  | ||||
|         // Create a dynamic string name - this will test memory management | ||||
|         std::stringstream ss; | ||||
|         ss << "thread_" << i << "_callback_" << j << "_id_" << callback_id; | ||||
|         std::string dynamic_name = ss.str(); | ||||
|  | ||||
|         ESP_LOGV(TAG, "Thread %d scheduling timeout with dynamic name: %s", i, dynamic_name.c_str()); | ||||
|  | ||||
|         // Capture necessary values for the lambda | ||||
|         auto *component = this; | ||||
|  | ||||
|         // Schedule with std::string name - this tests the string overload | ||||
|         // Use varying delays to stress the heap scheduler | ||||
|         uint32_t delay = 1 + (callback_id % 50); | ||||
|  | ||||
|         // Also test nested scheduling from callbacks | ||||
|         if (j % 10 == 0) { | ||||
|           // Every 10th callback schedules another callback | ||||
|           this->set_timeout(dynamic_name, delay, [component, i, j, callback_id]() { | ||||
|             component->executed_callbacks_.fetch_add(1); | ||||
|             ESP_LOGV(TAG, "Executed string-named callback %d (nested scheduler)", callback_id); | ||||
|  | ||||
|             // Schedule another timeout from within this callback with a new dynamic name | ||||
|             std::string nested_name = "nested_from_" + std::to_string(callback_id); | ||||
|             component->set_timeout(nested_name, 1, [component, callback_id]() { | ||||
|               ESP_LOGV(TAG, "Executed nested string-named callback from %d", callback_id); | ||||
|             }); | ||||
|           }); | ||||
|         } else { | ||||
|           // Regular callback | ||||
|           this->set_timeout(dynamic_name, delay, [component, i, j, callback_id]() { | ||||
|             component->executed_callbacks_.fetch_add(1); | ||||
|             ESP_LOGV(TAG, "Executed string-named callback %d", callback_id); | ||||
|           }); | ||||
|         } | ||||
|  | ||||
|         // Add some timing variations to increase race conditions | ||||
|         if (j % 5 == 0) { | ||||
|           std::this_thread::sleep_for(std::chrono::microseconds(100)); | ||||
|         } | ||||
|       } | ||||
|       ESP_LOGV(TAG, "Thread %d finished scheduling", i); | ||||
|     }); | ||||
|   } | ||||
|  | ||||
|   // Wait for all threads to complete scheduling | ||||
|   for (auto &t : threads) { | ||||
|     t.join(); | ||||
|   } | ||||
|  | ||||
|   auto end_time = std::chrono::steady_clock::now(); | ||||
|   auto thread_time = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count(); | ||||
|   ESP_LOGI(TAG, "All threads finished scheduling in %lldms. Created %d callbacks with dynamic names", thread_time, | ||||
|            this->total_callbacks_.load()); | ||||
|  | ||||
|   // Give some time for callbacks to execute | ||||
|   ESP_LOGI(TAG, "Waiting for callbacks to execute..."); | ||||
|  | ||||
|   // Schedule a final callback to signal completion | ||||
|   this->set_timeout("test_complete", 2000, [this]() { | ||||
|     ESP_LOGI(TAG, "String name stress test complete. Executed %d of %d callbacks", this->executed_callbacks_.load(), | ||||
|              this->total_callbacks_.load()); | ||||
|   }); | ||||
| } | ||||
|  | ||||
| }  // namespace scheduler_string_name_stress_component | ||||
| }  // namespace esphome | ||||
| @@ -0,0 +1,22 @@ | ||||
| #pragma once | ||||
|  | ||||
| #include "esphome/core/component.h" | ||||
| #include <atomic> | ||||
|  | ||||
| namespace esphome { | ||||
| namespace scheduler_string_name_stress_component { | ||||
|  | ||||
| class SchedulerStringNameStressComponent : public Component { | ||||
|  public: | ||||
|   void setup() override; | ||||
|   float get_setup_priority() const override { return setup_priority::LATE; } | ||||
|  | ||||
|   void run_string_name_stress_test(); | ||||
|  | ||||
|  private: | ||||
|   std::atomic<int> total_callbacks_{0}; | ||||
|   std::atomic<int> executed_callbacks_{0}; | ||||
| }; | ||||
|  | ||||
| }  // namespace scheduler_string_name_stress_component | ||||
| }  // namespace esphome | ||||
							
								
								
									
										38
									
								
								tests/integration/fixtures/scheduler_heap_stress.yaml
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										38
									
								
								tests/integration/fixtures/scheduler_heap_stress.yaml
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,38 @@ | ||||
| esphome: | ||||
|   name: scheduler-heap-stress-test | ||||
|  | ||||
| external_components: | ||||
|   - source: | ||||
|       type: local | ||||
|       path: EXTERNAL_COMPONENT_PATH | ||||
|     components: [scheduler_heap_stress_component] | ||||
|  | ||||
| host: | ||||
|  | ||||
| logger: | ||||
|   level: VERBOSE | ||||
|  | ||||
| scheduler_heap_stress_component: | ||||
|   id: heap_stress | ||||
|  | ||||
| api: | ||||
|   services: | ||||
|     - service: run_heap_stress_test | ||||
|       then: | ||||
|         - lambda: |- | ||||
|             id(heap_stress)->run_multi_thread_test(); | ||||
|  | ||||
| event: | ||||
|   - platform: template | ||||
|     name: "Test Complete" | ||||
|     id: test_complete | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "test_finished" | ||||
|   - platform: template | ||||
|     name: "Test Result" | ||||
|     id: test_result | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "passed" | ||||
|       - "failed" | ||||
							
								
								
									
										38
									
								
								tests/integration/fixtures/scheduler_rapid_cancellation.yaml
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										38
									
								
								tests/integration/fixtures/scheduler_rapid_cancellation.yaml
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,38 @@ | ||||
| esphome: | ||||
|   name: sched-rapid-cancel-test | ||||
|  | ||||
| external_components: | ||||
|   - source: | ||||
|       type: local | ||||
|       path: EXTERNAL_COMPONENT_PATH | ||||
|     components: [scheduler_rapid_cancellation_component] | ||||
|  | ||||
| host: | ||||
|  | ||||
| logger: | ||||
|   level: VERBOSE | ||||
|  | ||||
| scheduler_rapid_cancellation_component: | ||||
|   id: rapid_cancel | ||||
|  | ||||
| api: | ||||
|   services: | ||||
|     - service: run_rapid_cancellation_test | ||||
|       then: | ||||
|         - lambda: |- | ||||
|             id(rapid_cancel)->run_rapid_cancellation_test(); | ||||
|  | ||||
| event: | ||||
|   - platform: template | ||||
|     name: "Test Complete" | ||||
|     id: test_complete | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "test_finished" | ||||
|   - platform: template | ||||
|     name: "Test Result" | ||||
|     id: test_result | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "passed" | ||||
|       - "failed" | ||||
							
								
								
									
										38
									
								
								tests/integration/fixtures/scheduler_recursive_timeout.yaml
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										38
									
								
								tests/integration/fixtures/scheduler_recursive_timeout.yaml
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,38 @@ | ||||
| esphome: | ||||
|   name: sched-recursive-timeout | ||||
|  | ||||
| external_components: | ||||
|   - source: | ||||
|       type: local | ||||
|       path: EXTERNAL_COMPONENT_PATH | ||||
|     components: [scheduler_recursive_timeout_component] | ||||
|  | ||||
| host: | ||||
|  | ||||
| logger: | ||||
|   level: VERBOSE | ||||
|  | ||||
| scheduler_recursive_timeout_component: | ||||
|   id: recursive_timeout | ||||
|  | ||||
| api: | ||||
|   services: | ||||
|     - service: run_recursive_timeout_test | ||||
|       then: | ||||
|         - lambda: |- | ||||
|             id(recursive_timeout)->run_recursive_timeout_test(); | ||||
|  | ||||
| event: | ||||
|   - platform: template | ||||
|     name: "Test Complete" | ||||
|     id: test_complete | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "test_finished" | ||||
|   - platform: template | ||||
|     name: "Test Result" | ||||
|     id: test_result | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "passed" | ||||
|       - "failed" | ||||
| @@ -0,0 +1,23 @@ | ||||
| esphome: | ||||
|   name: sched-simul-callbacks-test | ||||
|  | ||||
| external_components: | ||||
|   - source: | ||||
|       type: local | ||||
|       path: EXTERNAL_COMPONENT_PATH | ||||
|     components: [scheduler_simultaneous_callbacks_component] | ||||
|  | ||||
| host: | ||||
|  | ||||
| logger: | ||||
|   level: INFO | ||||
|  | ||||
| scheduler_simultaneous_callbacks_component: | ||||
|   id: simultaneous_callbacks | ||||
|  | ||||
| api: | ||||
|   services: | ||||
|     - service: run_simultaneous_callbacks_test | ||||
|       then: | ||||
|         - lambda: |- | ||||
|             id(simultaneous_callbacks)->run_simultaneous_callbacks_test(); | ||||
							
								
								
									
										23
									
								
								tests/integration/fixtures/scheduler_string_lifetime.yaml
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										23
									
								
								tests/integration/fixtures/scheduler_string_lifetime.yaml
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,23 @@ | ||||
| esphome: | ||||
|   name: scheduler-string-lifetime-test | ||||
|  | ||||
| external_components: | ||||
|   - source: | ||||
|       type: local | ||||
|       path: EXTERNAL_COMPONENT_PATH | ||||
|     components: [scheduler_string_lifetime_component] | ||||
|  | ||||
| host: | ||||
|  | ||||
| logger: | ||||
|   level: DEBUG | ||||
|  | ||||
| scheduler_string_lifetime_component: | ||||
|   id: string_lifetime | ||||
|  | ||||
| api: | ||||
|   services: | ||||
|     - service: run_string_lifetime_test | ||||
|       then: | ||||
|         - lambda: |- | ||||
|             id(string_lifetime)->run_string_lifetime_test(); | ||||
							
								
								
									
										38
									
								
								tests/integration/fixtures/scheduler_string_name_stress.yaml
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										38
									
								
								tests/integration/fixtures/scheduler_string_name_stress.yaml
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,38 @@ | ||||
| esphome: | ||||
|   name: sched-string-name-stress | ||||
|  | ||||
| external_components: | ||||
|   - source: | ||||
|       type: local | ||||
|       path: EXTERNAL_COMPONENT_PATH | ||||
|     components: [scheduler_string_name_stress_component] | ||||
|  | ||||
| host: | ||||
|  | ||||
| logger: | ||||
|   level: VERBOSE | ||||
|  | ||||
| scheduler_string_name_stress_component: | ||||
|   id: string_stress | ||||
|  | ||||
| api: | ||||
|   services: | ||||
|     - service: run_string_name_stress_test | ||||
|       then: | ||||
|         - lambda: |- | ||||
|             id(string_stress)->run_string_name_stress_test(); | ||||
|  | ||||
| event: | ||||
|   - platform: template | ||||
|     name: "Test Complete" | ||||
|     id: test_complete | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "test_finished" | ||||
|   - platform: template | ||||
|     name: "Test Result" | ||||
|     id: test_result | ||||
|     device_class: button | ||||
|     event_types: | ||||
|       - "passed" | ||||
|       - "failed" | ||||
							
								
								
									
										148
									
								
								tests/integration/test_scheduler_heap_stress.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										148
									
								
								tests/integration/test_scheduler_heap_stress.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,148 @@ | ||||
| """Stress test for heap scheduler thread safety with multiple threads.""" | ||||
|  | ||||
| import asyncio | ||||
| from pathlib import Path | ||||
| import re | ||||
|  | ||||
| from aioesphomeapi import UserService | ||||
| import pytest | ||||
|  | ||||
| from .types import APIClientConnectedFactory, RunCompiledFunction | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_scheduler_heap_stress( | ||||
|     yaml_config: str, | ||||
|     run_compiled: RunCompiledFunction, | ||||
|     api_client_connected: APIClientConnectedFactory, | ||||
| ) -> None: | ||||
|     """Test that set_timeout/set_interval doesn't crash when called rapidly from multiple threads.""" | ||||
|  | ||||
|     # Get the absolute path to the external components directory | ||||
|     external_components_path = str( | ||||
|         Path(__file__).parent / "fixtures" / "external_components" | ||||
|     ) | ||||
|  | ||||
|     # Replace the placeholder in the YAML config with the actual path | ||||
|     yaml_config = yaml_config.replace( | ||||
|         "EXTERNAL_COMPONENT_PATH", external_components_path | ||||
|     ) | ||||
|  | ||||
|     # Create a future to signal test completion | ||||
|     loop = asyncio.get_event_loop() | ||||
|     test_complete_future: asyncio.Future[None] = loop.create_future() | ||||
|  | ||||
|     # Track executed timeouts/intervals and their order | ||||
|     executed_callbacks: set[int] = set() | ||||
|     thread_executions: dict[ | ||||
|         int, list[int] | ||||
|     ] = {}  # thread_id -> list of indices in execution order | ||||
|     callback_types: dict[int, str] = {}  # callback_id -> "timeout" or "interval" | ||||
|  | ||||
|     def on_log_line(line: str) -> None: | ||||
|         # Track all executed callbacks with thread and index info | ||||
|         match = re.search( | ||||
|             r"Executed (timeout|interval) (\d+) \(thread (\d+), index (\d+)\)", line | ||||
|         ) | ||||
|         if not match: | ||||
|             # Also check for the completion message | ||||
|             if "All threads finished" in line and "Created 1000 callbacks" in line: | ||||
|                 # Give scheduler some time to execute callbacks | ||||
|                 pass | ||||
|             return | ||||
|  | ||||
|         callback_type = match.group(1) | ||||
|         callback_id = int(match.group(2)) | ||||
|         thread_id = int(match.group(3)) | ||||
|         index = int(match.group(4)) | ||||
|  | ||||
|         # Only count each callback ID once (intervals might fire multiple times) | ||||
|         if callback_id not in executed_callbacks: | ||||
|             executed_callbacks.add(callback_id) | ||||
|             callback_types[callback_id] = callback_type | ||||
|  | ||||
|         # Track execution order per thread | ||||
|         if thread_id not in thread_executions: | ||||
|             thread_executions[thread_id] = [] | ||||
|  | ||||
|         # Only append if this is a new execution for this thread | ||||
|         if index not in thread_executions[thread_id]: | ||||
|             thread_executions[thread_id].append(index) | ||||
|  | ||||
|         # Check if we've executed all 1000 callbacks (0-999) | ||||
|         if len(executed_callbacks) >= 1000 and not test_complete_future.done(): | ||||
|             test_complete_future.set_result(None) | ||||
|  | ||||
|     async with ( | ||||
|         run_compiled(yaml_config, line_callback=on_log_line), | ||||
|         api_client_connected() as client, | ||||
|     ): | ||||
|         # Verify we can connect | ||||
|         device_info = await client.device_info() | ||||
|         assert device_info is not None | ||||
|         assert device_info.name == "scheduler-heap-stress-test" | ||||
|  | ||||
|         # List entities and services | ||||
|         _, services = await asyncio.wait_for( | ||||
|             client.list_entities_services(), timeout=5.0 | ||||
|         ) | ||||
|  | ||||
|         # Find our test service | ||||
|         run_stress_test_service: UserService | None = None | ||||
|         for service in services: | ||||
|             if service.name == "run_heap_stress_test": | ||||
|                 run_stress_test_service = service | ||||
|                 break | ||||
|  | ||||
|         assert run_stress_test_service is not None, ( | ||||
|             "run_heap_stress_test service not found" | ||||
|         ) | ||||
|  | ||||
|         # Call the run_heap_stress_test service to start the test | ||||
|         client.execute_service(run_stress_test_service, {}) | ||||
|  | ||||
|         # Wait for all callbacks to execute (should be quick, but give more time for scheduling) | ||||
|         try: | ||||
|             await asyncio.wait_for(test_complete_future, timeout=60.0) | ||||
|         except asyncio.TimeoutError: | ||||
|             # Report how many we got | ||||
|             pytest.fail( | ||||
|                 f"Stress test timed out. Only {len(executed_callbacks)} of " | ||||
|                 f"1000 callbacks executed. Missing IDs: " | ||||
|                 f"{sorted(set(range(1000)) - executed_callbacks)[:10]}..." | ||||
|             ) | ||||
|  | ||||
|         # Verify all callbacks executed | ||||
|         assert len(executed_callbacks) == 1000, ( | ||||
|             f"Expected 1000 callbacks, got {len(executed_callbacks)}" | ||||
|         ) | ||||
|  | ||||
|         # Verify we have all IDs from 0-999 | ||||
|         expected_ids = set(range(1000)) | ||||
|         missing_ids = expected_ids - executed_callbacks | ||||
|         assert not missing_ids, f"Missing callback IDs: {sorted(missing_ids)}" | ||||
|  | ||||
|         # Verify we have a mix of timeouts and intervals | ||||
|         timeout_count = sum(1 for t in callback_types.values() if t == "timeout") | ||||
|         interval_count = sum(1 for t in callback_types.values() if t == "interval") | ||||
|         assert timeout_count > 0, "No timeouts were executed" | ||||
|         assert interval_count > 0, "No intervals were executed" | ||||
|  | ||||
|         # Verify each thread executed callbacks | ||||
|         for thread_id, indices in thread_executions.items(): | ||||
|             assert len(indices) == 100, ( | ||||
|                 f"Thread {thread_id} executed {len(indices)} callbacks, expected 100" | ||||
|             ) | ||||
|  | ||||
|         # Verify that we executed a reasonable number of callbacks | ||||
|         assert timeout_count > 0, ( | ||||
|             f"Expected some timeout callbacks but got {timeout_count}" | ||||
|         ) | ||||
|         assert interval_count > 0, ( | ||||
|             f"Expected some interval callbacks but got {interval_count}" | ||||
|         ) | ||||
|         # Total should be 1000 callbacks | ||||
|         total_callbacks = timeout_count + interval_count | ||||
|         assert total_callbacks == 1000, ( | ||||
|             f"Expected 1000 total callbacks but got {total_callbacks}" | ||||
|         ) | ||||
							
								
								
									
										142
									
								
								tests/integration/test_scheduler_rapid_cancellation.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										142
									
								
								tests/integration/test_scheduler_rapid_cancellation.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,142 @@ | ||||
| """Rapid cancellation test - schedule and immediately cancel timeouts with string names.""" | ||||
|  | ||||
| import asyncio | ||||
| from pathlib import Path | ||||
| import re | ||||
|  | ||||
| from aioesphomeapi import UserService | ||||
| import pytest | ||||
|  | ||||
| from .types import APIClientConnectedFactory, RunCompiledFunction | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_scheduler_rapid_cancellation( | ||||
|     yaml_config: str, | ||||
|     run_compiled: RunCompiledFunction, | ||||
|     api_client_connected: APIClientConnectedFactory, | ||||
| ) -> None: | ||||
|     """Test rapid schedule/cancel cycles that might expose race conditions.""" | ||||
|  | ||||
|     # Get the absolute path to the external components directory | ||||
|     external_components_path = str( | ||||
|         Path(__file__).parent / "fixtures" / "external_components" | ||||
|     ) | ||||
|  | ||||
|     # Replace the placeholder in the YAML config with the actual path | ||||
|     yaml_config = yaml_config.replace( | ||||
|         "EXTERNAL_COMPONENT_PATH", external_components_path | ||||
|     ) | ||||
|  | ||||
|     # Create a future to signal test completion | ||||
|     loop = asyncio.get_event_loop() | ||||
|     test_complete_future: asyncio.Future[None] = loop.create_future() | ||||
|  | ||||
|     # Track test progress | ||||
|     test_stats = { | ||||
|         "log_count": 0, | ||||
|         "errors": [], | ||||
|         "summary_scheduled": None, | ||||
|         "final_scheduled": 0, | ||||
|         "final_executed": 0, | ||||
|         "final_implicit_cancellations": 0, | ||||
|     } | ||||
|  | ||||
|     def on_log_line(line: str) -> None: | ||||
|         # Count log lines | ||||
|         test_stats["log_count"] += 1 | ||||
|  | ||||
|         # Check for errors | ||||
|         if "ERROR" in line or "WARN" in line: | ||||
|             test_stats["errors"].append(line) | ||||
|  | ||||
|         # Parse summary statistics | ||||
|         if "All threads completed. Scheduled:" in line: | ||||
|             # Extract the scheduled count from the summary | ||||
|             if match := re.search(r"Scheduled: (\d+)", line): | ||||
|                 test_stats["summary_scheduled"] = int(match.group(1)) | ||||
|         elif "Total scheduled:" in line: | ||||
|             if match := re.search(r"Total scheduled: (\d+)", line): | ||||
|                 test_stats["final_scheduled"] = int(match.group(1)) | ||||
|         elif "Total executed:" in line: | ||||
|             if match := re.search(r"Total executed: (\d+)", line): | ||||
|                 test_stats["final_executed"] = int(match.group(1)) | ||||
|         elif "Implicit cancellations (replaced):" in line: | ||||
|             if match := re.search(r"Implicit cancellations \(replaced\): (\d+)", line): | ||||
|                 test_stats["final_implicit_cancellations"] = int(match.group(1)) | ||||
|  | ||||
|         # Check for crash indicators | ||||
|         if any( | ||||
|             indicator in line.lower() | ||||
|             for indicator in ["segfault", "abort", "assertion", "heap corruption"] | ||||
|         ): | ||||
|             if not test_complete_future.done(): | ||||
|                 test_complete_future.set_exception(Exception(f"Crash detected: {line}")) | ||||
|             return | ||||
|  | ||||
|         # Check for completion | ||||
|         if ( | ||||
|             "Rapid cancellation test complete" in line | ||||
|             and not test_complete_future.done() | ||||
|         ): | ||||
|             test_complete_future.set_result(None) | ||||
|  | ||||
|     async with ( | ||||
|         run_compiled(yaml_config, line_callback=on_log_line), | ||||
|         api_client_connected() as client, | ||||
|     ): | ||||
|         # Verify we can connect | ||||
|         device_info = await client.device_info() | ||||
|         assert device_info is not None | ||||
|         assert device_info.name == "sched-rapid-cancel-test" | ||||
|  | ||||
|         # List entities and services | ||||
|         _, services = await asyncio.wait_for( | ||||
|             client.list_entities_services(), timeout=5.0 | ||||
|         ) | ||||
|  | ||||
|         # Find our test service | ||||
|         run_test_service: UserService | None = None | ||||
|         for service in services: | ||||
|             if service.name == "run_rapid_cancellation_test": | ||||
|                 run_test_service = service | ||||
|                 break | ||||
|  | ||||
|         assert run_test_service is not None, ( | ||||
|             "run_rapid_cancellation_test service not found" | ||||
|         ) | ||||
|  | ||||
|         # Call the service to start the test | ||||
|         client.execute_service(run_test_service, {}) | ||||
|  | ||||
|         # Wait for test to complete with timeout | ||||
|         try: | ||||
|             await asyncio.wait_for(test_complete_future, timeout=10.0) | ||||
|         except asyncio.TimeoutError: | ||||
|             pytest.fail(f"Test timed out. Stats: {test_stats}") | ||||
|  | ||||
|         # Check for any errors | ||||
|         assert len(test_stats["errors"]) == 0, ( | ||||
|             f"Errors detected: {test_stats['errors']}" | ||||
|         ) | ||||
|  | ||||
|         # Check that we received log messages | ||||
|         assert test_stats["log_count"] > 0, "No log messages received" | ||||
|  | ||||
|         # Check the summary line to verify all threads scheduled their operations | ||||
|         assert test_stats["summary_scheduled"] == 400, ( | ||||
|             f"Expected summary to show 400 scheduled operations but got {test_stats['summary_scheduled']}" | ||||
|         ) | ||||
|  | ||||
|         # Check final statistics | ||||
|         assert test_stats["final_scheduled"] == 400, ( | ||||
|             f"Expected final stats to show 400 scheduled but got {test_stats['final_scheduled']}" | ||||
|         ) | ||||
|  | ||||
|         assert test_stats["final_executed"] == 10, ( | ||||
|             f"Expected final stats to show 10 executed but got {test_stats['final_executed']}" | ||||
|         ) | ||||
|  | ||||
|         assert test_stats["final_implicit_cancellations"] == 390, ( | ||||
|             f"Expected final stats to show 390 implicit cancellations but got {test_stats['final_implicit_cancellations']}" | ||||
|         ) | ||||
							
								
								
									
										101
									
								
								tests/integration/test_scheduler_recursive_timeout.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										101
									
								
								tests/integration/test_scheduler_recursive_timeout.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,101 @@ | ||||
| """Test for recursive timeout scheduling - scheduling timeouts from within timeout callbacks.""" | ||||
|  | ||||
| import asyncio | ||||
| from pathlib import Path | ||||
|  | ||||
| from aioesphomeapi import UserService | ||||
| import pytest | ||||
|  | ||||
| from .types import APIClientConnectedFactory, RunCompiledFunction | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_scheduler_recursive_timeout( | ||||
|     yaml_config: str, | ||||
|     run_compiled: RunCompiledFunction, | ||||
|     api_client_connected: APIClientConnectedFactory, | ||||
| ) -> None: | ||||
|     """Test that scheduling timeouts from within timeout callbacks works correctly.""" | ||||
|  | ||||
|     # Get the absolute path to the external components directory | ||||
|     external_components_path = str( | ||||
|         Path(__file__).parent / "fixtures" / "external_components" | ||||
|     ) | ||||
|  | ||||
|     # Replace the placeholder in the YAML config with the actual path | ||||
|     yaml_config = yaml_config.replace( | ||||
|         "EXTERNAL_COMPONENT_PATH", external_components_path | ||||
|     ) | ||||
|  | ||||
|     # Create a future to signal test completion | ||||
|     loop = asyncio.get_event_loop() | ||||
|     test_complete_future: asyncio.Future[None] = loop.create_future() | ||||
|  | ||||
|     # Track execution sequence | ||||
|     execution_sequence: list[str] = [] | ||||
|     expected_sequence = [ | ||||
|         "initial_timeout", | ||||
|         "nested_timeout_1", | ||||
|         "nested_timeout_2", | ||||
|         "test_complete", | ||||
|     ] | ||||
|  | ||||
|     def on_log_line(line: str) -> None: | ||||
|         # Track execution sequence | ||||
|         if "Executing initial timeout" in line: | ||||
|             execution_sequence.append("initial_timeout") | ||||
|         elif "Executing nested timeout 1" in line: | ||||
|             execution_sequence.append("nested_timeout_1") | ||||
|         elif "Executing nested timeout 2" in line: | ||||
|             execution_sequence.append("nested_timeout_2") | ||||
|         elif "Recursive timeout test complete" in line: | ||||
|             execution_sequence.append("test_complete") | ||||
|             if not test_complete_future.done(): | ||||
|                 test_complete_future.set_result(None) | ||||
|  | ||||
|     async with ( | ||||
|         run_compiled(yaml_config, line_callback=on_log_line), | ||||
|         api_client_connected() as client, | ||||
|     ): | ||||
|         # Verify we can connect | ||||
|         device_info = await client.device_info() | ||||
|         assert device_info is not None | ||||
|         assert device_info.name == "sched-recursive-timeout" | ||||
|  | ||||
|         # List entities and services | ||||
|         _, services = await asyncio.wait_for( | ||||
|             client.list_entities_services(), timeout=5.0 | ||||
|         ) | ||||
|  | ||||
|         # Find our test service | ||||
|         run_test_service: UserService | None = None | ||||
|         for service in services: | ||||
|             if service.name == "run_recursive_timeout_test": | ||||
|                 run_test_service = service | ||||
|                 break | ||||
|  | ||||
|         assert run_test_service is not None, ( | ||||
|             "run_recursive_timeout_test service not found" | ||||
|         ) | ||||
|  | ||||
|         # Call the service to start the test | ||||
|         client.execute_service(run_test_service, {}) | ||||
|  | ||||
|         # Wait for test to complete | ||||
|         try: | ||||
|             await asyncio.wait_for(test_complete_future, timeout=10.0) | ||||
|         except asyncio.TimeoutError: | ||||
|             pytest.fail( | ||||
|                 f"Recursive timeout test timed out. Got sequence: {execution_sequence}" | ||||
|             ) | ||||
|  | ||||
|         # Verify execution sequence | ||||
|         assert execution_sequence == expected_sequence, ( | ||||
|             f"Execution sequence mismatch. Expected {expected_sequence}, " | ||||
|             f"got {execution_sequence}" | ||||
|         ) | ||||
|  | ||||
|         # Verify we got exactly 4 events (Initial + Level 1 + Level 2 + Complete) | ||||
|         assert len(execution_sequence) == 4, ( | ||||
|             f"Expected 4 events but got {len(execution_sequence)}" | ||||
|         ) | ||||
							
								
								
									
										125
									
								
								tests/integration/test_scheduler_simultaneous_callbacks.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										125
									
								
								tests/integration/test_scheduler_simultaneous_callbacks.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,125 @@ | ||||
| """Simultaneous callbacks test - schedule many callbacks for the same time from multiple threads.""" | ||||
|  | ||||
| import asyncio | ||||
| from pathlib import Path | ||||
| import re | ||||
|  | ||||
| from aioesphomeapi import UserService | ||||
| import pytest | ||||
|  | ||||
| from .types import APIClientConnectedFactory, RunCompiledFunction | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_scheduler_simultaneous_callbacks( | ||||
|     yaml_config: str, | ||||
|     run_compiled: RunCompiledFunction, | ||||
|     api_client_connected: APIClientConnectedFactory, | ||||
| ) -> None: | ||||
|     """Test scheduling many callbacks for the exact same time from multiple threads.""" | ||||
|  | ||||
|     # Get the absolute path to the external components directory | ||||
|     external_components_path = str( | ||||
|         Path(__file__).parent / "fixtures" / "external_components" | ||||
|     ) | ||||
|  | ||||
|     # Replace the placeholder in the YAML config with the actual path | ||||
|     yaml_config = yaml_config.replace( | ||||
|         "EXTERNAL_COMPONENT_PATH", external_components_path | ||||
|     ) | ||||
|  | ||||
|     # Create a future to signal test completion | ||||
|     loop = asyncio.get_event_loop() | ||||
|     test_complete_future: asyncio.Future[None] = loop.create_future() | ||||
|  | ||||
|     # Track test progress | ||||
|     test_stats = { | ||||
|         "scheduled": 0, | ||||
|         "executed": 0, | ||||
|         "expected": 1000,  # 10 threads * 100 callbacks | ||||
|         "errors": [], | ||||
|     } | ||||
|  | ||||
|     def on_log_line(line: str) -> None: | ||||
|         # Track operations | ||||
|         if "Scheduled callback" in line: | ||||
|             test_stats["scheduled"] += 1 | ||||
|         elif "Callback executed" in line: | ||||
|             test_stats["executed"] += 1 | ||||
|         elif "ERROR" in line or "WARN" in line: | ||||
|             test_stats["errors"].append(line) | ||||
|  | ||||
|         # Check for crash indicators | ||||
|         if any( | ||||
|             indicator in line.lower() | ||||
|             for indicator in ["segfault", "abort", "assertion", "heap corruption"] | ||||
|         ): | ||||
|             if not test_complete_future.done(): | ||||
|                 test_complete_future.set_exception(Exception(f"Crash detected: {line}")) | ||||
|             return | ||||
|  | ||||
|         # Check for completion with final count | ||||
|         if "Final executed count:" in line: | ||||
|             # Extract number from log line like: "[07:59:47][I][simultaneous_callbacks:093]: Simultaneous callbacks test complete. Final executed count: 1000" | ||||
|             match = re.search(r"Final executed count:\s*(\d+)", line) | ||||
|             if match: | ||||
|                 test_stats["final_count"] = int(match.group(1)) | ||||
|  | ||||
|         # Check for completion | ||||
|         if ( | ||||
|             "Simultaneous callbacks test complete" in line | ||||
|             and not test_complete_future.done() | ||||
|         ): | ||||
|             test_complete_future.set_result(None) | ||||
|  | ||||
|     async with ( | ||||
|         run_compiled(yaml_config, line_callback=on_log_line), | ||||
|         api_client_connected() as client, | ||||
|     ): | ||||
|         # Verify we can connect | ||||
|         device_info = await client.device_info() | ||||
|         assert device_info is not None | ||||
|         assert device_info.name == "sched-simul-callbacks-test" | ||||
|  | ||||
|         # List entities and services | ||||
|         _, services = await asyncio.wait_for( | ||||
|             client.list_entities_services(), timeout=5.0 | ||||
|         ) | ||||
|  | ||||
|         # Find our test service | ||||
|         run_test_service: UserService | None = None | ||||
|         for service in services: | ||||
|             if service.name == "run_simultaneous_callbacks_test": | ||||
|                 run_test_service = service | ||||
|                 break | ||||
|  | ||||
|         assert run_test_service is not None, ( | ||||
|             "run_simultaneous_callbacks_test service not found" | ||||
|         ) | ||||
|  | ||||
|         # Call the service to start the test | ||||
|         client.execute_service(run_test_service, {}) | ||||
|  | ||||
|         # Wait for test to complete | ||||
|         try: | ||||
|             await asyncio.wait_for(test_complete_future, timeout=30.0) | ||||
|         except asyncio.TimeoutError: | ||||
|             pytest.fail(f"Simultaneous callbacks test timed out. Stats: {test_stats}") | ||||
|         except Exception as e: | ||||
|             pytest.fail(f"Test failed: {e}\nStats: {test_stats}") | ||||
|  | ||||
|         # Check for any errors | ||||
|         assert len(test_stats["errors"]) == 0, ( | ||||
|             f"Errors detected: {test_stats['errors']}" | ||||
|         ) | ||||
|  | ||||
|         # Verify all callbacks executed using the final count from C++ | ||||
|         final_count = test_stats.get("final_count", 0) | ||||
|         assert final_count == test_stats["expected"], ( | ||||
|             f"Expected {test_stats['expected']} callbacks, but only {final_count} executed" | ||||
|         ) | ||||
|  | ||||
|         # The final_count is the authoritative count from the C++ component | ||||
|         assert final_count == 1000, ( | ||||
|             f"Expected 1000 executed callbacks but got {final_count}" | ||||
|         ) | ||||
							
								
								
									
										130
									
								
								tests/integration/test_scheduler_string_lifetime.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										130
									
								
								tests/integration/test_scheduler_string_lifetime.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,130 @@ | ||||
| """String lifetime test - verify scheduler handles string destruction correctly.""" | ||||
|  | ||||
| import asyncio | ||||
| from pathlib import Path | ||||
| import re | ||||
|  | ||||
| from aioesphomeapi import UserService | ||||
| import pytest | ||||
|  | ||||
| from .types import APIClientConnectedFactory, RunCompiledFunction | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_scheduler_string_lifetime( | ||||
|     yaml_config: str, | ||||
|     run_compiled: RunCompiledFunction, | ||||
|     api_client_connected: APIClientConnectedFactory, | ||||
| ) -> None: | ||||
|     """Test that scheduler correctly handles string lifetimes when strings go out of scope.""" | ||||
|  | ||||
|     # Get the absolute path to the external components directory | ||||
|     external_components_path = str( | ||||
|         Path(__file__).parent / "fixtures" / "external_components" | ||||
|     ) | ||||
|  | ||||
|     # Replace the placeholder in the YAML config with the actual path | ||||
|     yaml_config = yaml_config.replace( | ||||
|         "EXTERNAL_COMPONENT_PATH", external_components_path | ||||
|     ) | ||||
|  | ||||
|     # Create a future to signal test completion | ||||
|     loop = asyncio.get_event_loop() | ||||
|     test_complete_future: asyncio.Future[None] = loop.create_future() | ||||
|  | ||||
|     # Track test progress | ||||
|     test_stats = { | ||||
|         "tests_passed": 0, | ||||
|         "tests_failed": 0, | ||||
|         "errors": [], | ||||
|         "use_after_free_detected": False, | ||||
|     } | ||||
|  | ||||
|     def on_log_line(line: str) -> None: | ||||
|         # Track test results from the C++ test output | ||||
|         if "Tests passed:" in line and "string_lifetime" in line: | ||||
|             # Extract the number from "Tests passed: 32" | ||||
|             match = re.search(r"Tests passed:\s*(\d+)", line) | ||||
|             if match: | ||||
|                 test_stats["tests_passed"] = int(match.group(1)) | ||||
|         elif "Tests failed:" in line and "string_lifetime" in line: | ||||
|             match = re.search(r"Tests failed:\s*(\d+)", line) | ||||
|             if match: | ||||
|                 test_stats["tests_failed"] = int(match.group(1)) | ||||
|         elif "ERROR" in line and "string_lifetime" in line: | ||||
|             test_stats["errors"].append(line) | ||||
|  | ||||
|         # Check for memory corruption indicators | ||||
|         if any( | ||||
|             indicator in line.lower() | ||||
|             for indicator in [ | ||||
|                 "use after free", | ||||
|                 "heap corruption", | ||||
|                 "segfault", | ||||
|                 "abort", | ||||
|                 "assertion", | ||||
|                 "sanitizer", | ||||
|                 "bad memory", | ||||
|                 "invalid pointer", | ||||
|             ] | ||||
|         ): | ||||
|             test_stats["use_after_free_detected"] = True | ||||
|             if not test_complete_future.done(): | ||||
|                 test_complete_future.set_exception( | ||||
|                     Exception(f"Memory corruption detected: {line}") | ||||
|                 ) | ||||
|             return | ||||
|  | ||||
|         # Check for completion | ||||
|         if "String lifetime tests complete" in line and not test_complete_future.done(): | ||||
|             test_complete_future.set_result(None) | ||||
|  | ||||
|     async with ( | ||||
|         run_compiled(yaml_config, line_callback=on_log_line), | ||||
|         api_client_connected() as client, | ||||
|     ): | ||||
|         # Verify we can connect | ||||
|         device_info = await client.device_info() | ||||
|         assert device_info is not None | ||||
|         assert device_info.name == "scheduler-string-lifetime-test" | ||||
|  | ||||
|         # List entities and services | ||||
|         _, services = await asyncio.wait_for( | ||||
|             client.list_entities_services(), timeout=5.0 | ||||
|         ) | ||||
|  | ||||
|         # Find our test service | ||||
|         run_test_service: UserService | None = None | ||||
|         for service in services: | ||||
|             if service.name == "run_string_lifetime_test": | ||||
|                 run_test_service = service | ||||
|                 break | ||||
|  | ||||
|         assert run_test_service is not None, ( | ||||
|             "run_string_lifetime_test service not found" | ||||
|         ) | ||||
|  | ||||
|         # Call the service to start the test | ||||
|         client.execute_service(run_test_service, {}) | ||||
|  | ||||
|         # Wait for test to complete | ||||
|         try: | ||||
|             await asyncio.wait_for(test_complete_future, timeout=30.0) | ||||
|         except asyncio.TimeoutError: | ||||
|             pytest.fail(f"String lifetime test timed out. Stats: {test_stats}") | ||||
|         except Exception as e: | ||||
|             pytest.fail(f"Test failed: {e}\nStats: {test_stats}") | ||||
|  | ||||
|         # Check for use-after-free | ||||
|         assert not test_stats["use_after_free_detected"], "Use-after-free detected!" | ||||
|  | ||||
|         # Check for any errors | ||||
|         assert test_stats["tests_failed"] == 0, f"Tests failed: {test_stats['errors']}" | ||||
|  | ||||
|         # Verify we had the expected number of passing tests and no failures | ||||
|         assert test_stats["tests_passed"] == 30, ( | ||||
|             f"Expected exactly 30 tests to pass, but got {test_stats['tests_passed']}" | ||||
|         ) | ||||
|         assert test_stats["tests_failed"] == 0, ( | ||||
|             f"Expected no test failures, but got {test_stats['tests_failed']} failures" | ||||
|         ) | ||||
							
								
								
									
										127
									
								
								tests/integration/test_scheduler_string_name_stress.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										127
									
								
								tests/integration/test_scheduler_string_name_stress.py
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,127 @@ | ||||
| """Stress test for heap scheduler with std::string names from multiple threads.""" | ||||
|  | ||||
| import asyncio | ||||
| from pathlib import Path | ||||
| import re | ||||
|  | ||||
| from aioesphomeapi import UserService | ||||
| import pytest | ||||
|  | ||||
| from .types import APIClientConnectedFactory, RunCompiledFunction | ||||
|  | ||||
|  | ||||
| @pytest.mark.asyncio | ||||
| async def test_scheduler_string_name_stress( | ||||
|     yaml_config: str, | ||||
|     run_compiled: RunCompiledFunction, | ||||
|     api_client_connected: APIClientConnectedFactory, | ||||
| ) -> None: | ||||
|     """Test that set_timeout/set_interval with std::string names doesn't crash when called from multiple threads.""" | ||||
|  | ||||
|     # Get the absolute path to the external components directory | ||||
|     external_components_path = str( | ||||
|         Path(__file__).parent / "fixtures" / "external_components" | ||||
|     ) | ||||
|  | ||||
|     # Replace the placeholder in the YAML config with the actual path | ||||
|     yaml_config = yaml_config.replace( | ||||
|         "EXTERNAL_COMPONENT_PATH", external_components_path | ||||
|     ) | ||||
|  | ||||
|     # Create a future to signal test completion | ||||
|     loop = asyncio.get_event_loop() | ||||
|     test_complete_future: asyncio.Future[None] = loop.create_future() | ||||
|  | ||||
|     # Track executed callbacks and any crashes | ||||
|     executed_callbacks: set[int] = set() | ||||
|     crash_detected = False | ||||
|     error_messages: list[str] = [] | ||||
|  | ||||
|     def on_log_line(line: str) -> None: | ||||
|         nonlocal crash_detected | ||||
|  | ||||
|         # Check for crash indicators | ||||
|         if any( | ||||
|             indicator in line.lower() | ||||
|             for indicator in [ | ||||
|                 "segfault", | ||||
|                 "abort", | ||||
|                 "assertion", | ||||
|                 "heap corruption", | ||||
|                 "use after free", | ||||
|             ] | ||||
|         ): | ||||
|             crash_detected = True | ||||
|             error_messages.append(line) | ||||
|             if not test_complete_future.done(): | ||||
|                 test_complete_future.set_exception(Exception(f"Crash detected: {line}")) | ||||
|             return | ||||
|  | ||||
|         # Track executed callbacks | ||||
|         match = re.search(r"Executed string-named callback (\d+)", line) | ||||
|         if match: | ||||
|             callback_id = int(match.group(1)) | ||||
|             executed_callbacks.add(callback_id) | ||||
|  | ||||
|         # Check for completion | ||||
|         if ( | ||||
|             "String name stress test complete" in line | ||||
|             and not test_complete_future.done() | ||||
|         ): | ||||
|             test_complete_future.set_result(None) | ||||
|  | ||||
|     async with ( | ||||
|         run_compiled(yaml_config, line_callback=on_log_line), | ||||
|         api_client_connected() as client, | ||||
|     ): | ||||
|         # Verify we can connect | ||||
|         device_info = await client.device_info() | ||||
|         assert device_info is not None | ||||
|         assert device_info.name == "sched-string-name-stress" | ||||
|  | ||||
|         # List entities and services | ||||
|         _, services = await asyncio.wait_for( | ||||
|             client.list_entities_services(), timeout=5.0 | ||||
|         ) | ||||
|  | ||||
|         # Find our test service | ||||
|         run_stress_test_service: UserService | None = None | ||||
|         for service in services: | ||||
|             if service.name == "run_string_name_stress_test": | ||||
|                 run_stress_test_service = service | ||||
|                 break | ||||
|  | ||||
|         assert run_stress_test_service is not None, ( | ||||
|             "run_string_name_stress_test service not found" | ||||
|         ) | ||||
|  | ||||
|         # Call the service to start the test | ||||
|         client.execute_service(run_stress_test_service, {}) | ||||
|  | ||||
|         # Wait for test to complete or crash | ||||
|         try: | ||||
|             await asyncio.wait_for(test_complete_future, timeout=30.0) | ||||
|         except asyncio.TimeoutError: | ||||
|             pytest.fail( | ||||
|                 f"String name stress test timed out. Executed {len(executed_callbacks)} callbacks. " | ||||
|                 f"This might indicate a deadlock." | ||||
|             ) | ||||
|         except Exception as e: | ||||
|             # A crash was detected | ||||
|             pytest.fail( | ||||
|                 f"Test failed due to crash: {e}\nError messages: {error_messages}" | ||||
|             ) | ||||
|  | ||||
|         # Verify no crashes occurred | ||||
|         assert not crash_detected, ( | ||||
|             f"Crash detected during test. Errors: {error_messages}" | ||||
|         ) | ||||
|  | ||||
|         # Verify we executed all 1000 callbacks (10 threads × 100 callbacks each) | ||||
|         assert len(executed_callbacks) == 1000, ( | ||||
|             f"Expected 1000 callbacks but got {len(executed_callbacks)}" | ||||
|         ) | ||||
|  | ||||
|         # Verify each callback ID was executed exactly once | ||||
|         for i in range(1000): | ||||
|             assert i in executed_callbacks, f"Callback {i} was not executed" | ||||
		Reference in New Issue
	
	Block a user