diff --git a/docker/MQTTManager/include/room/room.cpp b/docker/MQTTManager/include/room/room.cpp index e1f20bad..83c33e32 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 359c0237..883c2679 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 636e56e0..d2ee2d70 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 00000000..a44e7a38 --- /dev/null +++ b/docker/MQTTManager/tests/room_test.cpp @@ -0,0 +1,95 @@ +#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, 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()); + + 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 5930b3c3..7c6f6475 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;