From fc8f8ac1bb7be0f97ef49c00dd82b8622b5c5b6e Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Sat, 17 Oct 2026 13:57:09 -0700 Subject: [PATCH 1/2] MQTTManager: stop a room's status thread when the room is destroyed Each Room started a detached thread that waited on the room's own mutex and condition variable, and sent the room's status, forever. Nothing stopped it, so once a room was deleted and EntityManager dropped it, the thread kept using the freed Room: it either stayed blocked on a freed condition variable or, if a change was waiting to settle, went on to build and publish a status from freed memory. That could crash the manager at an unrelated point, or publish a stale state to the deleted room's topic just after ~Room() had cleared it. The thread is now a std::jthread whose waits take its stop token, and ~Room() stops and joins it before anything else is torn down. Stopping wakes the thread from either wait, so the join only waits for an update already being sent. The backoff used wait_for() returning no_timeout to tell that another change had arrived. The stop token overloads take a predicate instead, so changes are now counted and the backoff waits for the count to change. This also stops a spurious wakeup from restarting the backoff. EntityManager::load_rooms() now destroys removed rooms after releasing _rooms_mutex, so that the join never waits while holding it. If a room is ever destroyed on its own thread, which nothing does today, it logs and detaches instead of throwing from join(). Co-Authored-By: Claude Opus 5.5 --- .../include/entity_manager/entity_manager.cpp | 15 ++-- docker/MQTTManager/include/room/room.cpp | 45 ++++++++--- docker/MQTTManager/include/room/room.hpp | 14 +++- docker/MQTTManager/tests/main.cpp | 10 ++- docker/MQTTManager/tests/room_test.cpp | 78 +++++++++++++++++++ docker/MQTTManager/tests/test_helpers.hpp | 5 +- 6 files changed, 142 insertions(+), 25 deletions(-) create mode 100644 docker/MQTTManager/tests/room_test.cpp diff --git a/docker/MQTTManager/include/entity_manager/entity_manager.cpp b/docker/MQTTManager/include/entity_manager/entity_manager.cpp index a7b285c65..903523b1e 100644 --- a/docker/MQTTManager/include/entity_manager/entity_manager.cpp +++ b/docker/MQTTManager/include/entity_manager/entity_manager.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -120,14 +121,18 @@ void EntityManager::load_rooms() { auto room_ids = database_manager::database.select(&database_manager::Room::id, sqlite_orm::from()); SPDLOG_INFO("Loading {} rooms.", room_ids.size()); - // Remove room if it does no longer exist + // Remove room if it does no longer exist. The removed rooms are destroyed after _rooms_mutex is + // released, as destroying a room waits for its status update thread to stop. + std::vector> removed_rooms; { std::lock_guard mutex_guard(EntityManager::_rooms_mutex); - EntityManager::_rooms.erase(std::remove_if(EntityManager::_rooms.begin(), EntityManager::_rooms.end(), [&room_ids](auto room) { - return std::find_if(room_ids.begin(), room_ids.end(), [&room](auto id) { return id == room->get_id(); }) == room_ids.end(); - }), - EntityManager::_rooms.end()); + auto removed = std::stable_partition(EntityManager::_rooms.begin(), EntityManager::_rooms.end(), [&room_ids](auto room) { + return std::find_if(room_ids.begin(), room_ids.end(), [&room](auto id) { return id == room->get_id(); }) != room_ids.end(); + }); + removed_rooms.assign(std::make_move_iterator(removed), std::make_move_iterator(EntityManager::_rooms.end())); + EntityManager::_rooms.erase(removed, EntityManager::_rooms.end()); } + removed_rooms.clear(); // Cause existing room to reload config or add a new room if it does not exist. for (auto &room_id : room_ids) { diff --git a/docker/MQTTManager/include/room/room.cpp b/docker/MQTTManager/include/room/room.cpp index e1f20badd..83c33e323 100644 --- a/docker/MQTTManager/include/room/room.cpp +++ b/docker/MQTTManager/include/room/room.cpp @@ -36,14 +36,26 @@ Room::Room(uint32_t room_id) { this->_send_status_updates = true; CommandManager::attach_callback(boost::bind(&Room::command_callback, this, _1)); - if (!this->_update_room_state_thread.joinable()) { - SPDLOG_INFO("No thread to handle room status updates for room {}::{} starting...", this->_id, this->_name); - this->_update_room_state_thread = std::thread(&Room::_update_room_state, this); - this->_update_room_state_thread.detach(); - } + SPDLOG_INFO("No thread to handle room status updates for room {}::{} starting...", this->_id, this->_name); + // Through a lambda: std::jthread only passes its stop_token to a callable that takes it first. + this->_update_room_state_thread = std::jthread([this](std::stop_token stop_token) { + this->_update_room_state(stop_token); + }); } Room::~Room() { + // Stop the status update thread first, while everything it uses still exists. Stopping wakes it + // from either of its waits, so this only waits for an update already being sent. + this->_update_room_state_thread.request_stop(); + if (this->_update_room_state_thread.get_id() == std::this_thread::get_id()) { + // Nothing on the thread holds a reference to its own Room, so this should not happen. Joining + // would throw, so leave the thread to finish by itself. + SPDLOG_ERROR("Room {}::{} destroyed from its own status update thread, cannot wait for the thread to stop.", this->_id, this->_name); + this->_update_room_state_thread.detach(); + } else if (this->_update_room_state_thread.joinable()) { + this->_update_room_state_thread.join(); + } + CommandManager::detach_callback(boost::bind(&Room::command_callback, this, _1)); MQTT_Manager::clear_retain(this->_mqtt_state_topic); // Clear old state topic. SPDLOG_INFO("Room {}::{} destroyed.", this->_id, this->_name); @@ -280,6 +292,7 @@ void Room::page_changed_callback(RoomEntitiesPage *page) { { std::lock_guard mutex_guard(this->_status_update_mutex); this->_room_status_updated = false; + this->_room_update_count++; } SPDLOG_DEBUG("Sending notify of state update in room {}::{}", this->_id, this->_name); this->_room_update_condition_variable.notify_all(); @@ -360,20 +373,32 @@ void Room::command_callback(NSPanelMQTTManagerCommand &command) { } } -void Room::_update_room_state() { +void Room::_update_room_state(std::stop_token stop_token) { SPDLOG_INFO("Started thread to handle room status updates for room {}::{}", this->_id, this->_name); std::unique_lock lock(this->_status_update_mutex); for (;;) { // Wait for notification that a room has been updated - this->_room_update_condition_variable.wait(lock, [this]() { + this->_room_update_condition_variable.wait(lock, stop_token, [this]() { return !this->_room_status_updated; }); + if (stop_token.stop_requested()) { + return; + } // Wait until changes has settled as when a user changes light states in "All rooms" mode a burst of changes will occur from all rooms. uint32_t backoff_time = MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::ROOM_STATUS_BACKOFF_TIME); - while (this->_room_update_condition_variable.wait_for(lock, std::chrono::milliseconds(backoff_time)) == std::cv_status::no_timeout) { - // If we woke up via notify, we just loop and wait again. - // This is where the "debounce" happens. + for (;;) { + uint64_t seen_update_count = this->_room_update_count; + bool changed = this->_room_update_condition_variable.wait_for(lock, stop_token, std::chrono::milliseconds(backoff_time), [this, seen_update_count]() { + return this->_room_update_count != seen_update_count; + }); + if (stop_token.stop_requested()) { + return; + } + if (!changed) { + break; // A whole backoff time without changes. + } + // Another change arrived, wait a whole backoff time again. This is where the "debounce" happens. } SPDLOG_DEBUG("Room status backoff time reaced, sending update for room {}::{}.", this->_id, this->_name); this->_room_status_updated = true; diff --git a/docker/MQTTManager/include/room/room.hpp b/docker/MQTTManager/include/room/room.hpp index 359c0237c..883c26793 100644 --- a/docker/MQTTManager/include/room/room.hpp +++ b/docker/MQTTManager/include/room/room.hpp @@ -4,13 +4,17 @@ #include "entity/entity.hpp" #include "protobuf_nspanel.pb.h" #include "room/room_entities_page.hpp" +#include #include +#include #include #include #include #include #include +#include #include +#include #include class Room { @@ -131,13 +135,17 @@ class Room { private: /* * Run this in seperate thread to send room state updates with a backoff time to wait for changes to settle in order to minimise flickering. + * Returns once stop_token is stopped, which the destructor does before anything the thread uses is torn down. */ - void _update_room_state(); + void _update_room_state(std::stop_token stop_token); - std::thread _update_room_state_thread; + std::jthread _update_room_state_thread; std::mutex _status_update_mutex; - std::condition_variable _room_update_condition_variable; + std::condition_variable_any _room_update_condition_variable; std::atomic _room_status_updated = false; + // Counts change notifications, so that the backoff can tell a new change from the backoff time + // running out. Guarded by _status_update_mutex. + uint64_t _room_update_count = 0; /* * Create a protobuf room state object and send out to _mqtt_state_topic. diff --git a/docker/MQTTManager/tests/main.cpp b/docker/MQTTManager/tests/main.cpp index 636e56e01..d2ee2d70b 100644 --- a/docker/MQTTManager/tests/main.cpp +++ b/docker/MQTTManager/tests/main.cpp @@ -60,10 +60,12 @@ int main(int argc, char **argv) { database_manager::init(); int result = RUN_ALL_TESTS(); - // Skip static destructors. Each Room runs a detached thread that waits on the room's own mutex - // and condition variable forever; destroying EntityManager's static room list at exit pulls - // those out from under the threads and segfaults after every test has passed. The manager - // itself never exits this way. Results, including the JUnit XML, are written by now. + // Skip static destructors. EntityManager's static rooms and entities detach themselves from other + // static singletons (CommandManager, MQTT_Manager, the Home Assistant and OpenHAB managers, spdlog) + // when destroyed, and statics in different files are destroyed in no defined order. Some + // components also keep threads running that nothing stops. Exiting cleanly would need an explicit + // shutdown that tears entities down first. The manager itself never exits this way. Results, + // including the JUnit XML, are written by now. spdlog::default_logger()->flush(); std::fflush(nullptr); std::_Exit(result); diff --git a/docker/MQTTManager/tests/room_test.cpp b/docker/MQTTManager/tests/room_test.cpp new file mode 100644 index 000000000..c8740a7a8 --- /dev/null +++ b/docker/MQTTManager/tests/room_test.cpp @@ -0,0 +1,78 @@ +#include "test_helpers.hpp" +#include +#include +#include +#include +#include +#include +#include + +// Each Room runs a thread that sends its status once changes have settled. The thread must stop +// when the Room is destroyed, rather than go on using it. +class RoomStatusThreadTest : public nspm_test::SendCaptureTest { +protected: + // A Room row that EntityManager has not loaded, so that the test owns the only Room for it. + class ScopedRoomRow { + public: + explicit ScopedRoomRow(const std::string &friendly_name) { + database_manager::Room room; + room.friendly_name = friendly_name; + id = database_manager::database.insert(room); + } + + ~ScopedRoomRow() { + database_manager::database.remove(id); + } + + ScopedRoomRow(const ScopedRoomRow &) = delete; + ScopedRoomRow &operator=(const ScopedRoomRow &) = delete; + + int id; + }; + + static std::string state_topic(int room_id) { + return fmt::format("nspanel/mqttmanager_{}/room/{}/state", MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::MANAGER_ADDRESS), room_id); + } + + static std::chrono::milliseconds backoff_time() { + return std::chrono::milliseconds(MqttManagerConfig::get_setting_with_default(MQTT_MANAGER_SETTING::ROOM_STATUS_BACKOFF_TIME)); + } +}; + +TEST_F(RoomStatusThreadTest, sends_one_status_after_a_burst_of_changes) { + ScopedRoomRow row("Room thread test burst"); + auto room = std::make_unique(row.id); + + room->page_changed_callback(nullptr); + room->page_changed_callback(nullptr); + room->page_changed_callback(nullptr); + std::this_thread::sleep_for(backoff_time() * 4); + + EXPECT_EQ(nspm_test::mqtt_published_to(state_topic(row.id)).size(), 1); +} + +TEST_F(RoomStatusThreadTest, destroying_a_room_waiting_to_send_its_status_stops_its_thread) { + ScopedRoomRow row("Room thread test destroyed"); + auto room = std::make_unique(row.id); + + // The thread is now waiting for the change to settle, and would send the status once it has. + room->page_changed_callback(nullptr); + auto destroy_started = std::chrono::steady_clock::now(); + room.reset(); + auto destroy_time = std::chrono::steady_clock::now() - destroy_started; + + std::this_thread::sleep_for(backoff_time() * 4); + EXPECT_TRUE(nspm_test::mqtt_published_to(state_topic(row.id)).empty()); + // Stopping wakes the thread rather than waiting out the backoff. Generous, only to catch a hang. + EXPECT_LT(destroy_time, std::chrono::seconds(1)); +} + +TEST_F(RoomStatusThreadTest, a_room_removed_from_the_database_is_unloaded) { + int room_id = nspm_test::create_room("Room thread test removed"); + ASSERT_TRUE(EntityManager::get_room(room_id).has_value()); + + database_manager::database.remove(room_id); + EntityManager::load_rooms(); + + EXPECT_FALSE(EntityManager::get_room(room_id).has_value()); +} diff --git a/docker/MQTTManager/tests/test_helpers.hpp b/docker/MQTTManager/tests/test_helpers.hpp index 5930b3c37..7c6f6475b 100644 --- a/docker/MQTTManager/tests/test_helpers.hpp +++ b/docker/MQTTManager/tests/test_helpers.hpp @@ -110,9 +110,8 @@ class ScopedNSPanel { int id; }; -// Adds a Room row and loads it into the EntityManager. Rooms are never removed again: each -// Room starts a detached status thread that keeps using the Room, so destroying one is not -// safe. Call it with a name no other test uses. +// Adds a Room row and loads it into the EntityManager. The row is not removed again. Call it with +// a name no other test uses. inline int create_room(const std::string &friendly_name) { database_manager::Room room; room.friendly_name = friendly_name; From 188f33552478373bc4756d3f42d155672bc8357c Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Sun, 18 Oct 2026 07:43:40 -0700 Subject: [PATCH 2/2] MQTTManager: drop load_rooms() change, test destroying an idle room Revert load_rooms() to destroy removed rooms with erase/remove_if. Nothing on a room's status thread takes _rooms_mutex, so joining it under that lock cannot deadlock, and the extra code was not needed. Add a test that destroys a room whose status thread is waiting for the next change, which only the stop request can wake. Co-Authored-By: Claude Opus 5.5 --- .../include/entity_manager/entity_manager.cpp | 15 +++++---------- docker/MQTTManager/tests/room_test.cpp | 17 +++++++++++++++++ 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/docker/MQTTManager/include/entity_manager/entity_manager.cpp b/docker/MQTTManager/include/entity_manager/entity_manager.cpp index 903523b1e..a7b285c65 100644 --- a/docker/MQTTManager/include/entity_manager/entity_manager.cpp +++ b/docker/MQTTManager/include/entity_manager/entity_manager.cpp @@ -34,7 +34,6 @@ #include #include #include -#include #include #include #include @@ -121,18 +120,14 @@ void EntityManager::load_rooms() { auto room_ids = database_manager::database.select(&database_manager::Room::id, sqlite_orm::from()); SPDLOG_INFO("Loading {} rooms.", room_ids.size()); - // Remove room if it does no longer exist. The removed rooms are destroyed after _rooms_mutex is - // released, as destroying a room waits for its status update thread to stop. - std::vector> removed_rooms; + // Remove room if it does no longer exist { std::lock_guard mutex_guard(EntityManager::_rooms_mutex); - auto removed = std::stable_partition(EntityManager::_rooms.begin(), EntityManager::_rooms.end(), [&room_ids](auto room) { - return std::find_if(room_ids.begin(), room_ids.end(), [&room](auto id) { return id == room->get_id(); }) != room_ids.end(); - }); - removed_rooms.assign(std::make_move_iterator(removed), std::make_move_iterator(EntityManager::_rooms.end())); - EntityManager::_rooms.erase(removed, EntityManager::_rooms.end()); + EntityManager::_rooms.erase(std::remove_if(EntityManager::_rooms.begin(), EntityManager::_rooms.end(), [&room_ids](auto room) { + return std::find_if(room_ids.begin(), room_ids.end(), [&room](auto id) { return id == room->get_id(); }) == room_ids.end(); + }), + EntityManager::_rooms.end()); } - removed_rooms.clear(); // Cause existing room to reload config or add a new room if it does not exist. for (auto &room_id : room_ids) { diff --git a/docker/MQTTManager/tests/room_test.cpp b/docker/MQTTManager/tests/room_test.cpp index c8740a7a8..a44e7a38e 100644 --- a/docker/MQTTManager/tests/room_test.cpp +++ b/docker/MQTTManager/tests/room_test.cpp @@ -67,6 +67,23 @@ TEST_F(RoomStatusThreadTest, destroying_a_room_waiting_to_send_its_status_stops_ EXPECT_LT(destroy_time, std::chrono::seconds(1)); } +TEST_F(RoomStatusThreadTest, destroying_a_room_waiting_for_a_change_stops_its_thread) { + ScopedRoomRow row("Room thread test idle"); + auto room = std::make_unique(row.id); + + // Send one status, so that the thread is back waiting for the next change. + room->page_changed_callback(nullptr); + std::this_thread::sleep_for(backoff_time() * 4); + ASSERT_EQ(nspm_test::mqtt_published_to(state_topic(row.id)).size(), 1); + + // Nothing marks the room as changed, so only the stop request can end that wait. + auto destroy_started = std::chrono::steady_clock::now(); + room.reset(); + auto destroy_time = std::chrono::steady_clock::now() - destroy_started; + + EXPECT_LT(destroy_time, std::chrono::seconds(1)); +} + TEST_F(RoomStatusThreadTest, a_room_removed_from_the_database_is_unloaded) { int room_id = nspm_test::create_room("Room thread test removed"); ASSERT_TRUE(EntityManager::get_room(room_id).has_value());