feat(framework): Add scaling thread pool for the message handler (EVerest#1816)

A thread pool for handling the operation worker was added so that
commands and vars don't accumulate in the queue and cause delays which
can lead to timeouts in certain parts of the code.

Signed-off-by: Piet Gömpel <pietgoempel@gmail.com>
Signed-off-by: Martin Litre <mnlitre@gmail.com>

Author: mlitre

lib/everest/framework/BUILD.bazel

@@ -50,6 +50,7 @@ cc_library(
         "@rapidyaml//:rapidyaml",
         "//lib/everest/log:liblog",
         "//lib/everest/sqlite:everest-sqlite",
+        "//lib/everest/util:util",
         "@com_github_fmtlib_fmt//:fmt",
         "@com_github_nlohmann_json//:json",
         "@com_github_pboettch_json-schema-validator//:json-schema-validator",

lib/everest/framework/include/utils/message_handler.hpp

@@ -5,15 +5,17 @@
 
 #include <atomic>
 #include <condition_variable>
-#include <functional>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <queue>
 #include <string>
 #include <thread>
+#include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
+#include <everest/util/async/thread_pool_scaling.hpp>
 #include <utils/message_queue.hpp>
 #include <utils/types.hpp>
 
@@ -22,6 +24,10 @@ using CmdId = std::string;
 
 namespace Everest {
 
+constexpr int THREAD_POOL_SCALING_LATENCY_THRESHOLD_MS = 50;
+constexpr int THREAD_POOL_SCALING_LATENCY_THREAD_IDLE_TIMEOUT_S = 2;
+constexpr int THREAD_POOL_SCALING_MIN_THREAD_COUNT = 1;
+
 /// \brief Handles message dispatching and thread-safe queuing of different message types. This class uses two separate
 /// threads and message queues: one for operation messages (vars, cmds, errors, GetConfig, ModuleReady) and one for
 /// result messages (cmd results, GetConfig responses).
@@ -40,10 +46,14 @@ class MessageHandler {
     void register_handler(const std::string& topic, std::shared_ptr<TypedHandler> handler);
 
 private:
-    void run_operation_message_worker();
+    void run_operation_dispatcher();
     void run_result_message_worker();
     void run_external_mqtt_worker();
 
+    void dispatch_operation_message(ParsedMessage&& message);
+    void schedule_operation_message(ParsedMessage&& message);
+    void on_operation_message_done(const std::string& topic);
+
     void handle_operation_message(const std::string& topic, const json& payload);
     void handle_result_message(const std::string& topic, const json& payload);
 
@@ -69,20 +79,30 @@ class MessageHandler {
     void execute_single_handler(HandlerMap& handlers, const std::string& topic, ExecuteFn execute_fn);
 
     // Threads
-    std::thread
-        operation_worker_thread; // processes vars, commands, external MQTT, errors, GetConfig and ModuleReady messages
+    std::thread operation_dispatcher_thread; // processes vars, commands, external MQTT, errors, GetConfig and
+                                             // ModuleReady messages
     std::thread result_worker_thread;        // processes cmd results and GetConfig responses
     std::thread external_mqtt_worker_thread; // processes external MQTT messages
     std::thread ready_thread;                // runs the modules ready function
 
     // Queues and sync primitives
+    std::unique_ptr<everest::lib::util::thread_pool_scaling<
+        everest::lib::util::LatencyScaling<THREAD_POOL_SCALING_LATENCY_THRESHOLD_MS>>>
+        operation_thread_pool{std::make_unique<everest::lib::util::thread_pool_scaling<
+            everest::lib::util::LatencyScaling<THREAD_POOL_SCALING_LATENCY_THRESHOLD_MS>>>(
+            THREAD_POOL_SCALING_MIN_THREAD_COUNT, std::thread::hardware_concurrency(),
+            std::chrono::seconds(THREAD_POOL_SCALING_LATENCY_THREAD_IDLE_TIMEOUT_S))};
     std::queue<ParsedMessage> operation_message_queue;
     std::queue<ParsedMessage> result_message_queue;
     std::queue<ParsedMessage> external_mqtt_message_queue;
 
     std::mutex operation_queue_mutex;
     std::condition_variable operation_cv;
 
+    std::mutex operation_topic_state_mutex;
+    std::unordered_set<std::string> operation_topics_in_flight;
+    std::unordered_map<std::string, std::queue<ParsedMessage>> pending_operation_messages_by_topic;
+
     std::mutex result_queue_mutex;
     std::condition_variable result_cv;
 

lib/everest/framework/lib/CMakeLists.txt

@@ -77,6 +77,7 @@ target_link_libraries(framework
 
         everest::log
         everest::sqlite
+        everest::util
         ${STD_FILESYSTEM_COMPAT_LIB}
     PRIVATE
         ${EVEREST_FRAMEWORK_BOOST_SYSTEM_LINK_LIBRARY}

lib/everest/framework/lib/message_handler.cpp

@@ -6,6 +6,8 @@
 #include <everest/logging.hpp>
 #include <fmt/format.h>
 
+#include <optional>
+
 namespace Everest {
 
 namespace {
@@ -62,7 +64,7 @@ bool check_topic_matches(const std::string& full_topic, const std::string& wildc
 } // namespace
 
 MessageHandler::MessageHandler() {
-    operation_worker_thread = std::thread([this] { run_operation_message_worker(); });
+    operation_dispatcher_thread = std::thread([this] { run_operation_dispatcher(); });
     result_worker_thread = std::thread([this] { run_result_message_worker(); });
     external_mqtt_worker_thread = std::thread([this] { run_external_mqtt_worker(); });
 }
@@ -120,9 +122,13 @@ void MessageHandler::stop() {
     result_cv.notify_all();
     external_mqtt_cv.notify_all();
 
-    if (operation_worker_thread.joinable()) {
-        operation_worker_thread.join();
+    // Join the dispatcher first: it must not be able to call schedule_operation_message()
+    // (which dereferences operation_thread_pool) after the pool is destroyed.
+    if (operation_dispatcher_thread.joinable()) {
+        operation_dispatcher_thread.join();
     }
+    // The thread_pool destructor handles stopping and joining its workers.
+    operation_thread_pool.reset();
     if (result_worker_thread.joinable()) {
         result_worker_thread.join();
     }
@@ -134,7 +140,7 @@ void MessageHandler::stop() {
     }
 }
 
-void MessageHandler::run_operation_message_worker() {
+void MessageHandler::run_operation_dispatcher() {
     while (true) {
         std::unique_lock<std::mutex> lock(operation_queue_mutex);
         operation_cv.wait(lock, [this] { return !operation_message_queue.empty() || !running; });
@@ -145,9 +151,73 @@ void MessageHandler::run_operation_message_worker() {
         operation_message_queue.pop();
         lock.unlock();
 
-        handle_operation_message(message.topic, message.data);
+        dispatch_operation_message(std::move(message));
+    }
+    EVLOG_info << "Operation dispatcher thread stopped";
+}
+
+void MessageHandler::dispatch_operation_message(ParsedMessage&& message) {
+    {
+        std::lock_guard<std::mutex> lock(operation_topic_state_mutex);
+        if (operation_topics_in_flight.find(message.topic) != operation_topics_in_flight.end()) {
+            pending_operation_messages_by_topic[message.topic].push(std::move(message));
+            return;
+        }
+
+        operation_topics_in_flight.insert(message.topic);
+    }
+
+    schedule_operation_message(std::move(message));
+}
+
+// NOLINTNEXTLINE(misc-no-recursion)
+void MessageHandler::schedule_operation_message(ParsedMessage&& message) {
+    // NOLINTNEXTLINE(misc-no-recursion)
+    auto operation = [this, message = std::move(message)]() {
+        // Wrap in try-catch so that on_operation_message_done is always called: an exception in
+        // the handler must not leave the topic permanently stuck in operation_topics_in_flight,
+        // which would block all subsequent messages for that topic.
+        try {
+            handle_operation_message(message.topic, message.data);
+        } catch (const std::exception& e) {
+            EVLOG_error << "Exception while handling operation message on topic '" << message.topic
+                        << "': " << e.what();
+        } catch (...) {
+            EVLOG_error << "Unknown exception while handling operation message on topic '" << message.topic << "'";
+        }
+        on_operation_message_done(message.topic);
+    };
+
+    if (operation_thread_pool) {
+        operation_thread_pool->run(std::move(operation));
+    }
+}
+
+// NOLINTNEXTLINE(misc-no-recursion)
+void MessageHandler::on_operation_message_done(const std::string& topic) {
+    std::optional<ParsedMessage> next_message;
+    {
+        std::lock_guard<std::mutex> lock(operation_topic_state_mutex);
+        if (!running) {
+            // Shutting down: stop scheduling and release the in-flight slot.
+            operation_topics_in_flight.erase(topic);
+            return;
+        }
+        auto pending_it = pending_operation_messages_by_topic.find(topic);
+        if (pending_it != pending_operation_messages_by_topic.end() && !pending_it->second.empty()) {
+            next_message = std::move(pending_it->second.front());
+            pending_it->second.pop();
+
+            if (pending_it->second.empty()) {
+                pending_operation_messages_by_topic.erase(pending_it);
+            }
+        } else {
+            operation_topics_in_flight.erase(topic);
+            return;
+        }
     }
-    EVLOG_info << "Main worker thread stopped";
+
+    schedule_operation_message(std::move(*next_message));
 }
 
 void MessageHandler::run_result_message_worker() {

lib/everest/framework/schemas/config.yaml

@@ -231,5 +231,5 @@ properties:
     default: {}
     # don't allow arbitrary additional properties
     additionalProperties: false
-  
+
   x-module-layout: {}

lib/everest/framework/tests/CMakeLists.txt

@@ -16,6 +16,7 @@ target_sources(${TEST_TARGET_NAME} PRIVATE
     test_config_sqlite.cpp
     test_conversions.cpp
     test_filesystem_helpers.cpp
+    test_message_handler.cpp
     helpers.cpp
 )