MQTTManager: stop a room's status thread when the room is destroyed

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<std::mutex> 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<std::mutex> 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<uint32_t>(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;

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 <atomic>
 #include <chrono>
+#include <condition_variable>
 #include <expected>
 #include <memory>
 #include <mutex>
 #include <nlohmann/json_fwd.hpp>
 #include <protobuf/protobuf_general.pb.h>
+#include <stop_token>
 #include <string>
+#include <thread>
 #include <vector>
 
 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<bool> _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.

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);

docker/MQTTManager/tests/room_test.cpp

@@ -0,0 +1,95 @@
+#include "test_helpers.hpp"
+#include <chrono>
+#include <fmt/format.h>
+#include <gtest/gtest.h>
+#include <memory>
+#include <room/room.hpp>
+#include <string>
+#include <thread>
+
+// 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<database_manager::Room>(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<std::string>(MQTT_MANAGER_SETTING::MANAGER_ADDRESS), room_id);
+  }
+
+  static std::chrono::milliseconds backoff_time() {
+    return std::chrono::milliseconds(MqttManagerConfig::get_setting_with_default<uint32_t>(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<Room>(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<Room>(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<Room>(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<database_manager::Room>(room_id);
+  EntityManager::load_rooms();
+
+  EXPECT_FALSE(EntityManager::get_room(room_id).has_value());
+}

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;