mirror of
https://github.com/esphome/esphome.git
synced 2026-08-22 22:26:21 +00:00
[core] Fix wait_until crash when re-entered from its own continuation (#17571)
This commit is contained in:
@@ -502,6 +502,9 @@ template<typename... Ts> class WaitUntilAction : public Action<Ts...>, public Co
|
||||
|
||||
void stop() override {
|
||||
this->var_queue_.clear();
|
||||
// Tell any process_queue_() call further down the stack that the items it is
|
||||
// still holding were cancelled
|
||||
this->stop_generation_++;
|
||||
this->disable_loop();
|
||||
}
|
||||
|
||||
@@ -511,33 +514,57 @@ template<typename... Ts> class WaitUntilAction : public Action<Ts...>, public Co
|
||||
}
|
||||
|
||||
protected:
|
||||
using QueueItem = std::tuple<uint32_t, optional<uint32_t>, std::tuple<Ts...>>;
|
||||
|
||||
// Helper: Process queue, triggering completed items and removing them
|
||||
// Returns true if queue still has pending items
|
||||
bool process_queue_(uint32_t now) {
|
||||
// Process each queued wait_until and remove completed ones
|
||||
this->var_queue_.remove_if([&](auto &queued) {
|
||||
auto start = std::get<uint32_t>(queued);
|
||||
auto timeout = std::get<optional<uint32_t>>(queued);
|
||||
auto &var = std::get<std::tuple<Ts...>>(queued);
|
||||
// Completed items run the rest of the action chain synchronously, and that chain
|
||||
// can re-enter this same action (e.g. a script with mode: restart that executes
|
||||
// itself) and add to or clear var_queue_. Iterating the member list directly would
|
||||
// then corrupt it, so move it aside and iterate a local list instead.
|
||||
std::list<QueueItem> queue;
|
||||
queue.swap(this->var_queue_);
|
||||
std::list<QueueItem> pending;
|
||||
while (!queue.empty()) {
|
||||
auto it = queue.begin();
|
||||
auto start = std::get<uint32_t>(*it);
|
||||
auto timeout = std::get<optional<uint32_t>>(*it);
|
||||
|
||||
// Check if timeout has expired
|
||||
auto expired = timeout && (now - start) >= *timeout;
|
||||
|
||||
// Keep waiting if not expired and condition not met
|
||||
if (!expired && !this->condition_->check_tuple(var)) {
|
||||
return false;
|
||||
if (!expired && !this->condition_->check_tuple(std::get<std::tuple<Ts...>>(*it))) {
|
||||
pending.splice(pending.end(), queue, it);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Condition met or timed out - trigger next action
|
||||
this->play_next_tuple_(var);
|
||||
return true;
|
||||
});
|
||||
// Condition met or timed out - trigger the next action. Keep the item in a local
|
||||
// holder so its arguments stay valid while the chain runs, without any nested
|
||||
// process_queue_() call being able to see (and fire) it again.
|
||||
std::list<QueueItem> completed;
|
||||
completed.splice(completed.begin(), queue, it);
|
||||
uint8_t generation = this->stop_generation_;
|
||||
this->play_next_tuple_(std::get<std::tuple<Ts...>>(completed.front()));
|
||||
if (generation != this->stop_generation_) {
|
||||
// stop() ran inside the chain - the items still held locally were cancelled
|
||||
pending.clear();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Re-entrant continuations may have enqueued new waits into var_queue_; put the
|
||||
// older still-waiting items back in front of them to keep FIFO firing order
|
||||
this->var_queue_.splice(this->var_queue_.begin(), pending);
|
||||
|
||||
return !this->var_queue_.empty();
|
||||
}
|
||||
|
||||
Condition<Ts...> *condition_;
|
||||
std::list<std::tuple<uint32_t, optional<uint32_t>, std::tuple<Ts...>>> var_queue_{};
|
||||
std::list<QueueItem> var_queue_{};
|
||||
// Bumped by stop() so process_queue_() can detect a stop from inside play_next_tuple_()
|
||||
uint8_t stop_generation_{0};
|
||||
};
|
||||
|
||||
template<typename... Ts> class UpdateComponentAction : public Action<Ts...> {
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
esphome:
|
||||
name: wait-until-reentrant-restart
|
||||
|
||||
host:
|
||||
|
||||
api:
|
||||
actions:
|
||||
- action: start_self_restart
|
||||
then:
|
||||
- script.execute: retry_script
|
||||
- action: start_stop_during_wait
|
||||
then:
|
||||
- globals.set:
|
||||
id: gate_open
|
||||
value: 'false'
|
||||
# num 0 is a blocker: its condition never becomes true, so it is still
|
||||
# waiting (already checked and set aside) when num 1 stops the script -
|
||||
# it must be cancelled, not restored, so its timeout must never fire
|
||||
- script.execute:
|
||||
id: waiter
|
||||
num: 0
|
||||
- script.execute:
|
||||
id: waiter
|
||||
num: 1
|
||||
- script.execute:
|
||||
id: waiter
|
||||
num: 2
|
||||
- script.execute:
|
||||
id: waiter
|
||||
num: 3
|
||||
# Give all three instances time to queue in the same wait_until
|
||||
- delay: 100ms
|
||||
- globals.set:
|
||||
id: gate_open
|
||||
value: 'true'
|
||||
- delay: 200ms
|
||||
- logger.log: "stop test complete"
|
||||
|
||||
logger:
|
||||
level: DEBUG
|
||||
|
||||
globals:
|
||||
- id: attempt
|
||||
type: int
|
||||
initial_value: '0'
|
||||
- id: gate_open
|
||||
type: bool
|
||||
initial_value: 'false'
|
||||
|
||||
script:
|
||||
# Self-restart retry pattern: when the wait_until times out, the rest of the
|
||||
# script runs synchronously from inside the wait queue processing and restarts
|
||||
# this same script - re-entering the same WaitUntilAction while it is still
|
||||
# processing its queue. This used to corrupt the queue and crash.
|
||||
- id: retry_script
|
||||
mode: restart
|
||||
then:
|
||||
- wait_until:
|
||||
condition:
|
||||
lambda: 'return false;'
|
||||
timeout: 20ms
|
||||
- lambda: |-
|
||||
id(attempt) += 1;
|
||||
ESP_LOGD("test", "attempt %d done", id(attempt));
|
||||
- if:
|
||||
condition:
|
||||
lambda: 'return id(attempt) < 5;'
|
||||
then:
|
||||
- script.execute: retry_script
|
||||
else:
|
||||
- logger.log: "retry test complete"
|
||||
|
||||
# Parallel waiters all queued in the same wait_until; the first one to pass the
|
||||
# gate stops the script from its continuation, cancelling the other waiters
|
||||
# while the queue is still being processed.
|
||||
- id: waiter
|
||||
mode: parallel
|
||||
parameters:
|
||||
num: int
|
||||
then:
|
||||
- wait_until:
|
||||
condition:
|
||||
lambda: 'return num != 0 && id(gate_open);'
|
||||
timeout: 1s
|
||||
- lambda: 'ESP_LOGD("test", "gate passed %d", num);'
|
||||
- script.stop: waiter
|
||||
@@ -0,0 +1,89 @@
|
||||
"""Integration test for wait_until queue reentrancy.
|
||||
|
||||
When a wait_until completes, the rest of the action chain runs synchronously
|
||||
from inside the wait queue processing. That chain can re-enter the very same
|
||||
WaitUntilAction - for example a script with mode: restart that executes itself
|
||||
as a retry pattern, or a waiter that stops its own script. Both used to mutate
|
||||
the std::list while it was being iterated, corrupting it and crashing the
|
||||
device (Guru Meditation StoreProhibited in _M_transfer).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wait_until_reentrant_restart(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
api_client_connected: APIClientConnectedFactory,
|
||||
) -> None:
|
||||
"""Test that re-entering a wait_until from its own continuation is safe."""
|
||||
retry_complete = asyncio.Event()
|
||||
stop_complete = asyncio.Event()
|
||||
|
||||
attempt_pattern = re.compile(r"attempt (\d+) done")
|
||||
gate_pattern = re.compile(r"gate passed (\d+)")
|
||||
|
||||
attempts: list[int] = []
|
||||
gate_passed: list[int] = []
|
||||
|
||||
def check_output(line: str) -> None:
|
||||
"""Check log output for expected messages."""
|
||||
if mo := attempt_pattern.search(line):
|
||||
attempts.append(int(mo.group(1)))
|
||||
elif mo := gate_pattern.search(line):
|
||||
gate_passed.append(int(mo.group(1)))
|
||||
elif "retry test complete" in line:
|
||||
retry_complete.set()
|
||||
elif "stop test complete" in line:
|
||||
stop_complete.set()
|
||||
|
||||
async with (
|
||||
run_compiled(yaml_config, line_callback=check_output),
|
||||
api_client_connected() as client,
|
||||
):
|
||||
device_info = await client.device_info()
|
||||
assert device_info is not None
|
||||
assert device_info.name == "wait-until-reentrant-restart"
|
||||
|
||||
_, services = await client.list_entities_services()
|
||||
self_restart_service = next(
|
||||
(s for s in services if s.name == "start_self_restart"), None
|
||||
)
|
||||
assert self_restart_service is not None, "start_self_restart not found"
|
||||
stop_service = next(
|
||||
(s for s in services if s.name == "start_stop_during_wait"), None
|
||||
)
|
||||
assert stop_service is not None, "start_stop_during_wait not found"
|
||||
|
||||
# Scenario 1: the wait_until timeout continuation restarts its own
|
||||
# script five times, re-entering the same wait_until each time.
|
||||
await client.execute_service(self_restart_service, {})
|
||||
try:
|
||||
await asyncio.wait_for(retry_complete.wait(), timeout=10.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(f"Self-restart retry did not finish. Attempts: {attempts}")
|
||||
assert attempts == [1, 2, 3, 4, 5], attempts
|
||||
|
||||
# Scenario 2: the first waiter through the gate stops the script while
|
||||
# the other waiters are still queued in the same wait_until; both the
|
||||
# not-yet-checked waiters (2, 3) and the already-checked still-waiting
|
||||
# blocker (0) must be cancelled, not fired.
|
||||
await client.execute_service(stop_service, {})
|
||||
try:
|
||||
await asyncio.wait_for(stop_complete.wait(), timeout=10.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(f"Stop-during-wait did not finish. Gate passed: {gate_passed}")
|
||||
assert gate_passed == [1], gate_passed
|
||||
|
||||
# If the cancelled blocker had been kept, its 1s wait_until timeout
|
||||
# would still fire - give it the chance and check it stays silent.
|
||||
await asyncio.sleep(1.5)
|
||||
assert gate_passed == [1], gate_passed
|
||||
Reference in New Issue
Block a user