From 7f6348e46d0bb76fd790c05efdff6e550afac8da Mon Sep 17 00:00:00 2001 From: Fernandez Date: Thu, 6 Aug 2026 09:54:39 +0800 Subject: [PATCH] fix(http): harden thread pool lifecycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 校验线程池初始化参数并事务式创建、启动工作线程 - 在失败和关闭路径完整回收调度器、事件和线程资源 - 增加优先级分配、初始化回滚与重试生命周期测试 Closes #90 Signed-off-by: Fernandez --- src/network/http/HttpServer.cc | 13 +- src/network/http/HttpServerThreadPool.cc | 157 ++++++++------ src/network/http/HttpServerThreadPool.h | 12 +- test/test_http_server_lifetime.cc | 257 +++++++++++++++++++++++ 4 files changed, 371 insertions(+), 68 deletions(-) diff --git a/src/network/http/HttpServer.cc b/src/network/http/HttpServer.cc index 33b51f741..bf86124e9 100644 --- a/src/network/http/HttpServer.cc +++ b/src/network/http/HttpServer.cc @@ -14,6 +14,7 @@ #include #include #include +#include #include #include #include @@ -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; } diff --git a/src/network/http/HttpServerThreadPool.cc b/src/network/http/HttpServerThreadPool.cc index 265ae266d..105e65b8c 100644 --- a/src/network/http/HttpServerThreadPool.cc +++ b/src/network/http/HttpServerThreadPool.cc @@ -2,7 +2,13 @@ #include "network/http/HttpServerThreadPool.h" +#include +#include +#include #include +#include +#include +#include #include "network/http/HttpCommon.h" #include "network/http/HttpServerThread.h" @@ -10,104 +16,137 @@ #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 kPriority0Interfaces = {"dologin", "resetsystem"}; + constexpr std::array 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(name, server, factory()); + if (thread_num < static_cast(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(thread_num); + std::vector> 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(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(msg.GetData()); - if (!ptask) { - return nIdx; +std::optional HttpServerThreadPool::MsgInPrioIndex(const cosmo::MsgEnvelope& msg) const { + const auto* task = static_cast(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::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::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 diff --git a/src/network/http/HttpServerThreadPool.h b/src/network/http/HttpServerThreadPool.h index c91878cb5..119996832 100644 --- a/src/network/http/HttpServerThreadPool.h +++ b/src/network/http/HttpServerThreadPool.h @@ -1,9 +1,10 @@ #pragma once #include +#include #include #include -#include +#include #include #include "network/msg/MsgEnvelope.h" @@ -20,7 +21,7 @@ class HttpServerThreadPool { using DispatcherFactory = std::function()>; - // 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 MsgInPrioIndex(const cosmo::MsgEnvelope& msg) const; + std::vector> msg_handler_threads_; std::atomic is_accepting_{false}; - int thread_num_ = 4; - int cur_thread_idx_ = -1; - std::vector prio0_interface_; - std::vector prio1_interface_; }; } // namespace cosmo::network::http diff --git a/test/test_http_server_lifetime.cc b/test/test_http_server_lifetime.cc index 78a607e0b..d9ee7c294 100644 --- a/test/test_http_server_lifetime.cc +++ b/test/test_http_server_lifetime.cc @@ -2,6 +2,8 @@ #include #include +#include +#include #include #include #include @@ -10,6 +12,8 @@ #include #include #include +#include +#include #include #include #include @@ -17,6 +21,7 @@ #include "catch_amalgamated.hpp" #include "network/http/HttpServer.h" +#include "network/http/HttpServerThreadPool.h" #include "nlohmann/json.hpp" #include "util/ErrorCode.h" #include "util/IRequestDispatcher.h" @@ -141,6 +146,102 @@ namespace { std::shared_ptr state_; }; + struct DispatcherProbe { + std::atomic live_count{0}; + std::atomic dispatch_count{0}; + std::array dispatch_count_by_instance{}; + }; + + class CountingDispatcher final : public cosmo::IRequestDispatcher { + public: + explicit CountingDispatcher(std::shared_ptr probe, + std::optional instance_index = std::nullopt) + : probe_(std::move(probe)), instance_index_(instance_index) { + probe_->live_count.fetch_add(1, std::memory_order_relaxed); + } + + ~CountingDispatcher() override { + probe_->live_count.fetch_sub(1, std::memory_order_relaxed); + } + + bool SupportsRoute(const std::string& /*interface*/) override { + return true; + } + + cosmo::RequestAdmission InspectRequest(cosmo::RequestDispatchContext& context, + bool /*require_known_route*/) override { + context.principal = "thread-pool-test"; + return cosmo::RequestAdmission::kAllowed; + } + + bool DispatchRequest(const cosmo::RequestDispatchContext& /*context*/, const std::string& /*body*/, + std::string& response) override { + probe_->dispatch_count.fetch_add(1, std::memory_order_relaxed); + if (instance_index_ && *instance_index_ < probe_->dispatch_count_by_instance.size()) { + ++probe_->dispatch_count_by_instance[*instance_index_]; + } + response = R"({"ok":true})"; + return true; + } + + private: + std::shared_ptr probe_; + std::optional instance_index_; + }; + + class CopyThrowingDispatcherFactory { + public: + CopyThrowingDispatcherFactory(std::shared_ptr probe, + std::shared_ptr> should_throw) + : probe_(std::move(probe)), should_throw_(std::move(should_throw)) {} + + CopyThrowingDispatcherFactory(const CopyThrowingDispatcherFactory& other) + : probe_(other.probe_), should_throw_(other.should_throw_) { + if (should_throw_->load(std::memory_order_relaxed)) { + throw std::runtime_error("dispatcher factory copy failure"); + } + } + + CopyThrowingDispatcherFactory(CopyThrowingDispatcherFactory&&) noexcept = default; + CopyThrowingDispatcherFactory& operator=(const CopyThrowingDispatcherFactory&) = delete; + CopyThrowingDispatcherFactory& operator=(CopyThrowingDispatcherFactory&&) = delete; + + std::unique_ptr operator()() const { + return std::make_unique(probe_); + } + + private: + std::shared_ptr probe_; + std::shared_ptr> should_throw_; + }; + + HttpServerThreadPool::DispatcherFactory MakeCountingDispatcherFactory( + const std::shared_ptr& probe, std::size_t* call_count = nullptr) { + return [probe, call_count]() -> std::unique_ptr { + std::optional instance_index; + if (call_count != nullptr) { + instance_index = (*call_count)++; + } + return std::make_unique(probe, instance_index); + }; + } + + HttpServerCallbacks MakeThreadPoolTestCallbacks() { + return { + []() { return std::string("/tmp"); }, + []() { return std::string("/tmp"); }, + []() { return std::string("/tmp"); }, + }; + } + + cosmo::MsgEnvelope MakeHttpRequestMessage(std::string interface) { + auto task = std::make_unique(); + task->request_time = std::chrono::steady_clock::now(); + task->interface = std::move(interface); + task->mtk = "thread-pool-token"; + return {static_cast(InnerMsgId::kHttpReq), std::move(task)}; + } + class ReleaseGuard { public: explicit ReleaseGuard(std::shared_ptr state) : state_(std::move(state)) {} @@ -347,6 +448,162 @@ namespace { } // namespace +TEST_CASE("HttpServerThreadPool rejects invalid initialization inputs", + "[http-server][thread-pool][lifecycle]") { + HttpServer server; + HttpServerThreadPool thread_pool; + auto probe = std::make_shared(); + std::size_t call_count = 0; + const auto valid_factory = MakeCountingDispatcherFactory(probe, &call_count); + + for (const int thread_count : {-1, 0, 1, 2}) { + CAPTURE(thread_count); + CHECK_FALSE(thread_pool.Initialize(thread_count, &server, valid_factory)); + } + CHECK(call_count == 0); + CHECK_FALSE(thread_pool.Initialize(3, nullptr, valid_factory)); + CHECK(call_count == 0); + CHECK_FALSE(thread_pool.Initialize(3, &server, HttpServerThreadPool::DispatcherFactory{})); + CHECK(call_count == 0); + + CHECK_NOTHROW(thread_pool.Uninitialize()); + CHECK_NOTHROW(thread_pool.Uninitialize()); + CHECK(probe->live_count.load(std::memory_order_relaxed) == 0); +} + +TEST_CASE("HttpServerThreadPool rolls back dispatcher factory failures", + "[http-server][thread-pool][lifecycle]") { + for (const bool should_throw : {false, true}) { + for (std::size_t failure_index = 0; failure_index < 3; ++failure_index) { + CAPTURE(should_throw, failure_index); + HttpServer server; + HttpServerThreadPool thread_pool; + auto probe = std::make_shared(); + std::size_t call_count = 0; + HttpServerThreadPool::DispatcherFactory failing_factory = + [probe, &call_count, failure_index, + should_throw]() -> std::unique_ptr { + const auto current_index = call_count++; + if (current_index == failure_index) { + if (should_throw) { + throw std::runtime_error("dispatcher factory failure"); + } + return nullptr; + } + return std::make_unique(probe); + }; + + CHECK_FALSE(thread_pool.Initialize(3, &server, std::move(failing_factory))); + CHECK(call_count == failure_index + 1); + CHECK(probe->live_count.load(std::memory_order_relaxed) == 0); + auto rejected_message = MakeHttpRequestMessage("/normal"); + CHECK(thread_pool.PutMsg(std::move(rejected_message)) == -1); + CHECK_NOTHROW(thread_pool.Uninitialize()); + + REQUIRE(thread_pool.Initialize(3, &server, MakeCountingDispatcherFactory(probe))); + CHECK(probe->live_count.load(std::memory_order_relaxed) == 3); + thread_pool.Uninitialize(); + CHECK(probe->live_count.load(std::memory_order_relaxed) == 0); + } + } +} + +TEST_CASE("HttpServerThreadPool preserves active workers and drains accepted requests", + "[http-server][thread-pool][lifecycle]") { + constexpr std::array kInterfaces = {"/normal", "/api/dologin", "/api/ResetSystem", + "/api/ThreadDebugInfo", "/api/QueryDeviceInfo"}; + + for (const int thread_count : {3, 4}) { + CAPTURE(thread_count); + HttpServer server; + HttpServerThreadPool thread_pool; + auto probe = std::make_shared(); + std::size_t call_count = 0; + const auto factory = MakeCountingDispatcherFactory(probe, &call_count); + + REQUIRE(thread_pool.Initialize(thread_count, &server, factory)); + CHECK(probe->live_count.load(std::memory_order_relaxed) == thread_count); + CHECK(call_count == static_cast(thread_count)); + + CHECK_FALSE(thread_pool.Initialize(thread_count, &server, factory)); + CHECK(call_count == static_cast(thread_count)); + + for (const auto* interface : kInterfaces) { + auto message = MakeHttpRequestMessage(interface); + REQUIRE(thread_pool.PutMsg(std::move(message)) >= 0); + } + + CHECK_NOTHROW(thread_pool.Uninitialize()); + CHECK(probe->dispatch_count.load(std::memory_order_relaxed) == static_cast(kInterfaces.size())); + CHECK(probe->dispatch_count_by_instance[0] == 2); + CHECK(probe->dispatch_count_by_instance[1] == 2); + CHECK(probe->dispatch_count_by_instance[2] == 1); + if (thread_count == 4) { + CHECK(probe->dispatch_count_by_instance[3] == 0); + } + CHECK(probe->live_count.load(std::memory_order_relaxed) == 0); + CHECK_NOTHROW(thread_pool.Uninitialize()); + } +} + +TEST_CASE("HttpServer releases resources after worker dispatcher construction fails", + "[http-server][thread-pool][lifecycle]") { + auto port = FindAvailablePort(); + REQUIRE(port != 0); + + HttpServer server; + auto failing_probe = std::make_shared(); + std::size_t call_count = 0; + constexpr std::size_t kFailureCall = 2; + HttpServer::DispatcherFactory failing_factory = + [failing_probe, &call_count]() -> std::unique_ptr { + const auto current_call = call_count++; + if (current_call == kFailureCall) { + throw std::runtime_error("worker dispatcher construction failure"); + } + return std::make_unique(failing_probe); + }; + + CHECK_FALSE( + server.Initialize("127.0.0.1", port, std::move(failing_factory), MakeThreadPoolTestCallbacks())); + CHECK(call_count == kFailureCall + 1); + CHECK(failing_probe->live_count.load(std::memory_order_relaxed) == 0); + + auto valid_probe = std::make_shared(); + REQUIRE(server.Initialize("127.0.0.1", port, MakeCountingDispatcherFactory(valid_probe), + MakeThreadPoolTestCallbacks())); + CHECK(valid_probe->live_count.load(std::memory_order_relaxed) == 5); + CHECK_NOTHROW(server.UnInitialize()); + CHECK(valid_probe->live_count.load(std::memory_order_relaxed) == 0); + CHECK_NOTHROW(server.UnInitialize()); +} + +TEST_CASE("HttpServer releases resources when the worker dispatcher factory copy throws", + "[http-server][thread-pool][lifecycle]") { + auto port = FindAvailablePort(); + REQUIRE(port != 0); + + HttpServer server; + auto failing_probe = std::make_shared(); + auto should_throw = std::make_shared>(false); + HttpServer::DispatcherFactory failing_factory = + CopyThrowingDispatcherFactory(failing_probe, should_throw); + should_throw->store(true, std::memory_order_relaxed); + + bool is_initialized = true; + REQUIRE_NOTHROW(is_initialized = server.Initialize("127.0.0.1", port, std::move(failing_factory), + MakeThreadPoolTestCallbacks())); + CHECK_FALSE(is_initialized); + CHECK(failing_probe->live_count.load(std::memory_order_relaxed) == 0); + + auto valid_probe = std::make_shared(); + REQUIRE(server.Initialize("127.0.0.1", port, MakeCountingDispatcherFactory(valid_probe), + MakeThreadPoolTestCallbacks())); + CHECK(valid_probe->live_count.load(std::memory_order_relaxed) == 5); + CHECK_NOTHROW(server.UnInitialize()); + CHECK(valid_probe->live_count.load(std::memory_order_relaxed) == 0); +} + TEST_CASE("HttpServer keeps libevent requests on the event thread during shutdown", "[http-server][thread]") { ScopedSignalIgnore ignore_sigpipe(SIGPIPE); auto state = std::make_shared();