fix(http): harden thread pool lifecycle

src/network/http/HttpServer.cc

@@ -14,6 +14,7 @@
 #include <charconv>
 #include <cstdio>
 #include <cstring>
+#include <exception>
 #include <filesystem>
 #include <limits>
 #include <memory>
@@ -648,8 +649,16 @@ bool HttpServer::Initialize(const std::string& host_ip, int port, DispatcherFact
         return false;
     }
 
-    constexpr int kThreadNum = 4;
-    if (!thread_pool_.Initialize(kThreadNum, this, dispatcher_factory_)) {
+    constexpr int kThreadNum        = 4;
+    bool is_thread_pool_initialized = false;
+    try {
+        is_thread_pool_initialized = thread_pool_.Initialize(kThreadNum, this, dispatcher_factory_);
+    } catch (const std::exception& e) {
+        LOG_ERRO("Cannot initialize HTTP request thread pool: {}", e.what());
+    } catch (...) {
+        LOG_ERRO("{}", "Cannot initialize HTTP request thread pool: unknown exception");
+    }
+    if (!is_thread_pool_initialized) {
         CleanupEventResources();
         return false;
     }

src/network/http/HttpServerThreadPool.cc

@@ -2,112 +2,151 @@
 
 #include "network/http/HttpServerThreadPool.h"
 
+#include <array>
+#include <cstddef>
+#include <exception>
 #include <limits>
+#include <string>
+#include <utility>
+#include <vector>
 
 #include "network/http/HttpCommon.h"
 #include "network/http/HttpServerThread.h"
 #include "util/Log.h"
 #include "util/StringUtil.h"
 
 namespace cosmo::network::http {
+namespace {
 
-HttpServerThreadPool::HttpServerThreadPool() : thread_num_(4), cur_thread_idx_(-1) {}
+    constexpr std::size_t kPriority0WorkerIndex = 0;
+    constexpr std::size_t kPriority1WorkerIndex = 1;
+    constexpr std::size_t kNormalWorkerBegin    = 2;
+    constexpr std::size_t kMinimumThreadCount   = kNormalWorkerBegin + 1;
+
+    constexpr std::array<const char*, 2> kPriority0Interfaces = {"dologin", "resetsystem"};
+    constexpr std::array<const char*, 2> kPriority1Interfaces = {"threaddebuginfo", "querydeviceinfo"};
+
+}  // namespace
+
+HttpServerThreadPool::HttpServerThreadPool() = default;
 
 HttpServerThreadPool::~HttpServerThreadPool() {
     Uninitialize();
 }
 
 bool HttpServerThreadPool::Initialize(int thread_num, HttpServer* server, DispatcherFactory factory) {
-    if (thread_num <= 0) {
+    if (!msg_handler_threads_.empty()) {
+        LOG_ERRO("{}", "HttpServerThreadPool is already initialized");
         return false;
     }
-    is_accepting_ = false;
-    thread_num_   = thread_num;
-    msg_handler_threads_.resize(thread_num_);
-    for (int idx = 0; idx < thread_num; ++idx) {
-        char name[64] = {0};
-        snprintf(name, sizeof(name), "MsgHanderThread_%d", idx);
-        msg_handler_threads_[idx] = std::make_unique<MsgHanderThread>(name, server, factory());
+    if (thread_num < static_cast<int>(kMinimumThreadCount)) {
+        LOG_ERRO("HttpServerThreadPool requires at least {} handler threads, got {}", kMinimumThreadCount,
+                 thread_num);
+        return false;
     }
-
-    for (int idx = 0; idx < thread_num; ++idx) {
-        if (!msg_handler_threads_[idx]->start()) {
-            LOG_ERRO("HttpServerThreadPool failed to start handler thread {}", idx);
-            Uninitialize();
-            return false;
-        }
+    if (server == nullptr) {
+        LOG_ERRO("{}", "HttpServerThreadPool requires a valid HTTP server");
+        return false;
+    }
+    if (!factory) {
+        LOG_ERRO("{}", "HttpServerThreadPool dispatcher factory is not configured");
+        return false;
     }
 
-    prio0_interface_.clear();
-    prio1_interface_.clear();
+    is_accepting_.store(false, std::memory_order_release);
+    const auto handler_count = static_cast<std::size_t>(thread_num);
+    std::vector<std::unique_ptr<MsgHanderThread>> candidate_threads;
+    std::size_t handler_index = 0;
+
+    try {
+        candidate_threads.reserve(handler_count);
+        for (; handler_index < handler_count; ++handler_index) {
+            auto dispatcher = factory();
+            if (!dispatcher) {
+                LOG_ERRO("HttpServerThreadPool dispatcher factory returned null for handler {}",
+                         handler_index);
+                return false;
+            }
+
+            auto name = std::string("MsgHanderThread_") + std::to_string(handler_index);
+            candidate_threads.emplace_back(
+                std::make_unique<MsgHanderThread>(name, server, std::move(dispatcher)));
+        }
 
-    // Priority 0: restart, reset etc.
-    prio0_interface_.push_back(cosmo::util::ToLower("dologin"));
-    prio0_interface_.push_back(cosmo::util::ToLower("ResetSystem"));
+        for (handler_index = 0; handler_index < handler_count; ++handler_index) {
+            if (!candidate_threads[handler_index]->start()) {
+                LOG_ERRO("HttpServerThreadPool failed to start handler thread {}", handler_index);
+                return false;
+            }
+        }
+    } catch (const std::exception& ex) {
+        LOG_ERRO("HttpServerThreadPool initialization failed near handler {}: {}", handler_index, ex.what());
+        return false;
+    } catch (...) {
+        LOG_ERRO("HttpServerThreadPool initialization failed near handler {} with an unknown exception",
+                 handler_index);
+        return false;
+    }
 
-    // Priority 1: non-blocking or debug
-    prio1_interface_.push_back(cosmo::util::ToLower("ThreadDebugInfo"));
-    prio1_interface_.push_back(cosmo::util::ToLower("QueryDeviceInfo"));
-    is_accepting_ = true;
+    msg_handler_threads_.swap(candidate_threads);
+    is_accepting_.store(true, std::memory_order_release);
     return true;
 }
 
 void HttpServerThreadPool::Uninitialize() {
-    is_accepting_ = false;
-    if (!msg_handler_threads_.empty()) {
-        for (int idx = 0; idx < thread_num_; ++idx) {
-            msg_handler_threads_[idx]->DrainAndStop();
-        }
-
-        msg_handler_threads_.clear();
+    is_accepting_.store(false, std::memory_order_release);
+    for (auto& handler_thread : msg_handler_threads_) {
+        handler_thread->DrainAndStop();
     }
+    msg_handler_threads_.clear();
 }
 
-int HttpServerThreadPool::MsgInPrioIndex(cosmo::MsgEnvelope& msg) {
-    int nIdx    = -1;
-    auto* ptask = static_cast<HttpReqTask*>(msg.GetData());
-    if (!ptask) {
-        return nIdx;
+std::optional<std::size_t> HttpServerThreadPool::MsgInPrioIndex(const cosmo::MsgEnvelope& msg) const {
+    const auto* task = static_cast<const HttpReqTask*>(msg.GetData());
+    if (task == nullptr) {
+        return std::nullopt;
     }
 
-    auto interface = cosmo::util::ToLower(ptask->interface);
-    for (auto& prioInterface : prio0_interface_) {
-        if (std::string::npos != interface.find(prioInterface)) {
-            return 0;
+    const auto interface = cosmo::util::ToLower(task->interface);
+    for (const auto* priority_interface : kPriority0Interfaces) {
+        if (interface.find(priority_interface) != std::string::npos) {
+            return kPriority0WorkerIndex;
         }
     }
-    for (auto& prioInterface : prio1_interface_) {
-        if (std::string::npos != interface.find(prioInterface)) {
-            return 1;
+    for (const auto* priority_interface : kPriority1Interfaces) {
+        if (interface.find(priority_interface) != std::string::npos) {
+            return kPriority1WorkerIndex;
         }
     }
-    return nIdx;
+    return std::nullopt;
 }
 
 int HttpServerThreadPool::PutMsg(cosmo::MsgEnvelope&& msg) {
-    if (!is_accepting_ || msg_handler_threads_.empty())
+    if (!is_accepting_.load(std::memory_order_acquire) || msg_handler_threads_.size() < kMinimumThreadCount) {
         return -1;
+    }
 
-    size_t minMsgCount = std::numeric_limits<size_t>::max();
-    int nIdx           = MsgInPrioIndex(msg);
-    if (nIdx < 0) {
-        for (int idx = 2; idx < thread_num_; ++idx) {
-            auto msgCount = msg_handler_threads_[idx]->MsgCount();
-            if (minMsgCount > msgCount) {
-                minMsgCount = msgCount;
-                nIdx        = idx;
+    std::size_t handler_index = kNormalWorkerBegin;
+    std::size_t min_msg_count = std::numeric_limits<std::size_t>::max();
+    if (const auto priority_index = MsgInPrioIndex(msg)) {
+        handler_index = *priority_index;
+        min_msg_count = msg_handler_threads_[handler_index]->MsgCount();
+    } else {
+        for (std::size_t index = kNormalWorkerBegin; index < msg_handler_threads_.size(); ++index) {
+            const auto msg_count = msg_handler_threads_[index]->MsgCount();
+            if (min_msg_count > msg_count) {
+                min_msg_count = msg_count;
+                handler_index = index;
             }
 
-            if (0 == minMsgCount) {
+            if (min_msg_count == 0) {
                 break;
             }
         }
-    } else {
-        minMsgCount = msg_handler_threads_[nIdx]->MsgCount();
     }
 
-    LOG_INFO("PutMsg To Http Pool {}, This Pool Have {} Tasks in Queue", nIdx, minMsgCount);
-    return msg_handler_threads_[nIdx]->Put(std::move(msg));
+    LOG_INFO("PutMsg To Http Pool {}, This Pool Have {} Tasks in Queue", handler_index, min_msg_count);
+    return msg_handler_threads_[handler_index]->Put(std::move(msg));
 }
 
 }  // namespace cosmo::network::http

src/network/http/HttpServerThreadPool.h

@@ -1,9 +1,10 @@
 #pragma once
 
 #include <atomic>
+#include <cstddef>
 #include <functional>
 #include <memory>
-#include <string>
+#include <optional>
 #include <vector>
 
 #include "network/msg/MsgEnvelope.h"
@@ -20,7 +21,7 @@ class HttpServerThreadPool {
 
     using DispatcherFactory = std::function<std::unique_ptr<cosmo::IRequestDispatcher>()>;
 
-    // Initialize thread pool
+    // Initialize at least three workers. HttpServer serializes lifecycle calls with PutMsg().
     bool Initialize(int thread_num, HttpServer* server, DispatcherFactory factory);
 
     // Shutdown thread pool
@@ -30,13 +31,10 @@ class HttpServerThreadPool {
     int PutMsg(cosmo::MsgEnvelope&& msg);
 
 private:
-    int MsgInPrioIndex(cosmo::MsgEnvelope& msg);
+    std::optional<std::size_t> MsgInPrioIndex(const cosmo::MsgEnvelope& msg) const;
+
     std::vector<std::unique_ptr<MsgHanderThread>> msg_handler_threads_;
     std::atomic<bool> is_accepting_{false};
-    int thread_num_     = 4;
-    int cur_thread_idx_ = -1;
-    std::vector<std::string> prio0_interface_;
-    std::vector<std::string> prio1_interface_;
 };
 
 }  // namespace cosmo::network::http