diff --git a/tests/integration/fixtures/external_components/scheduler_heap_stress_component/__init__.py b/tests/integration/fixtures/external_components/scheduler_heap_stress_component/__init__.py new file mode 100644 index 0000000000..4540fa5667 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_heap_stress_component/__init__.py @@ -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) diff --git a/tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.cpp b/tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.cpp new file mode 100644 index 0000000000..2bb5147b07 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.cpp @@ -0,0 +1,104 @@ +#include "heap_scheduler_stress_component.h" +#include "esphome/core/log.h" +#include +#include +#include +#include +#include + +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 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(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 \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.h b/tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.h new file mode 100644 index 0000000000..36b55741af --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_heap_stress_component/heap_scheduler_stress_component.h @@ -0,0 +1,22 @@ +#pragma once + +#include "esphome/core/component.h" +#include + +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 total_callbacks_{0}; + std::atomic executed_callbacks_{0}; +}; + +} // namespace scheduler_heap_stress_component +} // namespace esphome \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/__init__.py b/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/__init__.py new file mode 100644 index 0000000000..0bb784e74e --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/__init__.py @@ -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) diff --git a/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.cpp b/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.cpp new file mode 100644 index 0000000000..210576e613 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.cpp @@ -0,0 +1,77 @@ +#include "rapid_cancellation_component.h" +#include "esphome/core/log.h" +#include +#include +#include +#include +#include + +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 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 diff --git a/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.h b/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.h new file mode 100644 index 0000000000..fdc1401940 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_rapid_cancellation_component/rapid_cancellation_component.h @@ -0,0 +1,22 @@ +#pragma once + +#include "esphome/core/component.h" +#include + +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 total_scheduled_{0}; + std::atomic total_executed_{0}; +}; + +} // namespace scheduler_rapid_cancellation_component +} // namespace esphome \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/__init__.py b/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/__init__.py new file mode 100644 index 0000000000..4e847a6fdb --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/__init__.py @@ -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) diff --git a/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.cpp b/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.cpp new file mode 100644 index 0000000000..48b33513f2 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.cpp @@ -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 \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.h b/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.h new file mode 100644 index 0000000000..e654353a1a --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_recursive_timeout_component/recursive_timeout_component.h @@ -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 \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/__init__.py b/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/__init__.py new file mode 100644 index 0000000000..bb1d560ad3 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/__init__.py @@ -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) diff --git a/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.cpp b/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.cpp new file mode 100644 index 0000000000..20dbc050e9 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.cpp @@ -0,0 +1,109 @@ +#include "simultaneous_callbacks_component.h" +#include "esphome/core/log.h" +#include +#include +#include +#include + +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 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 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 \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.h b/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.h new file mode 100644 index 0000000000..4dcc29d5b5 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_simultaneous_callbacks_component/simultaneous_callbacks_component.h @@ -0,0 +1,24 @@ +#pragma once + +#include "esphome/core/component.h" +#include + +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 total_scheduled_{0}; + std::atomic total_executed_{0}; + std::atomic callbacks_at_once_{0}; + std::atomic max_concurrent_{0}; +}; + +} // namespace scheduler_simultaneous_callbacks_component +} // namespace esphome \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/__init__.py b/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/__init__.py new file mode 100644 index 0000000000..3f29a839ef --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/__init__.py @@ -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) diff --git a/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.cpp b/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.cpp new file mode 100644 index 0000000000..7cc9d81bb0 --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.cpp @@ -0,0 +1,233 @@ +#include "string_lifetime_component.h" +#include "esphome/core/log.h" +#include +#include +#include + +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 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("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 \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.h b/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.h new file mode 100644 index 0000000000..fce075f31f --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_string_lifetime_component/string_lifetime_component.h @@ -0,0 +1,29 @@ +#pragma once + +#include "esphome/core/component.h" +#include +#include + +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 \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/__init__.py b/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/__init__.py new file mode 100644 index 0000000000..6cc564395c --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/__init__.py @@ -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) diff --git a/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.cpp b/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.cpp new file mode 100644 index 0000000000..f6f602a7bd --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.cpp @@ -0,0 +1,110 @@ +#include "string_name_stress_component.h" +#include "esphome/core/log.h" +#include +#include +#include +#include +#include +#include + +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 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(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 \ No newline at end of file diff --git a/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.h b/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.h new file mode 100644 index 0000000000..ac0020cdad --- /dev/null +++ b/tests/integration/fixtures/external_components/scheduler_string_name_stress_component/string_name_stress_component.h @@ -0,0 +1,22 @@ +#pragma once + +#include "esphome/core/component.h" +#include + +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 total_callbacks_{0}; + std::atomic executed_callbacks_{0}; +}; + +} // namespace scheduler_string_name_stress_component +} // namespace esphome \ No newline at end of file diff --git a/tests/integration/fixtures/scheduler_heap_stress.yaml b/tests/integration/fixtures/scheduler_heap_stress.yaml new file mode 100644 index 0000000000..d4d340b68b --- /dev/null +++ b/tests/integration/fixtures/scheduler_heap_stress.yaml @@ -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" diff --git a/tests/integration/fixtures/scheduler_rapid_cancellation.yaml b/tests/integration/fixtures/scheduler_rapid_cancellation.yaml new file mode 100644 index 0000000000..4824654c5c --- /dev/null +++ b/tests/integration/fixtures/scheduler_rapid_cancellation.yaml @@ -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" diff --git a/tests/integration/fixtures/scheduler_recursive_timeout.yaml b/tests/integration/fixtures/scheduler_recursive_timeout.yaml new file mode 100644 index 0000000000..f1168802f6 --- /dev/null +++ b/tests/integration/fixtures/scheduler_recursive_timeout.yaml @@ -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" diff --git a/tests/integration/fixtures/scheduler_simultaneous_callbacks.yaml b/tests/integration/fixtures/scheduler_simultaneous_callbacks.yaml new file mode 100644 index 0000000000..446ee7fdc0 --- /dev/null +++ b/tests/integration/fixtures/scheduler_simultaneous_callbacks.yaml @@ -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(); diff --git a/tests/integration/fixtures/scheduler_string_lifetime.yaml b/tests/integration/fixtures/scheduler_string_lifetime.yaml new file mode 100644 index 0000000000..a16f46f144 --- /dev/null +++ b/tests/integration/fixtures/scheduler_string_lifetime.yaml @@ -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(); diff --git a/tests/integration/fixtures/scheduler_string_name_stress.yaml b/tests/integration/fixtures/scheduler_string_name_stress.yaml new file mode 100644 index 0000000000..d1ef55c8d5 --- /dev/null +++ b/tests/integration/fixtures/scheduler_string_name_stress.yaml @@ -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" diff --git a/tests/integration/test_scheduler_heap_stress.py b/tests/integration/test_scheduler_heap_stress.py new file mode 100644 index 0000000000..d5f03462fd --- /dev/null +++ b/tests/integration/test_scheduler_heap_stress.py @@ -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}" + ) diff --git a/tests/integration/test_scheduler_rapid_cancellation.py b/tests/integration/test_scheduler_rapid_cancellation.py new file mode 100644 index 0000000000..9c5ed4bb6e --- /dev/null +++ b/tests/integration/test_scheduler_rapid_cancellation.py @@ -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']}" + ) diff --git a/tests/integration/test_scheduler_recursive_timeout.py b/tests/integration/test_scheduler_recursive_timeout.py new file mode 100644 index 0000000000..acd03215d1 --- /dev/null +++ b/tests/integration/test_scheduler_recursive_timeout.py @@ -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)}" + ) diff --git a/tests/integration/test_scheduler_simultaneous_callbacks.py b/tests/integration/test_scheduler_simultaneous_callbacks.py new file mode 100644 index 0000000000..de5ea601d6 --- /dev/null +++ b/tests/integration/test_scheduler_simultaneous_callbacks.py @@ -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}" + ) diff --git a/tests/integration/test_scheduler_string_lifetime.py b/tests/integration/test_scheduler_string_lifetime.py new file mode 100644 index 0000000000..3b79fc8b70 --- /dev/null +++ b/tests/integration/test_scheduler_string_lifetime.py @@ -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" + ) diff --git a/tests/integration/test_scheduler_string_name_stress.py b/tests/integration/test_scheduler_string_name_stress.py new file mode 100644 index 0000000000..a51a915300 --- /dev/null +++ b/tests/integration/test_scheduler_string_name_stress.py @@ -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"