diff --git a/be/src/http/ev_http_server.cpp b/be/src/http/ev_http_server.cpp index 457cf0e0322ba5..ceb53df9180b03 100644 --- a/be/src/http/ev_http_server.cpp +++ b/be/src/http/ev_http_server.cpp @@ -117,16 +117,14 @@ void EvHttpServer::start() { .set_min_threads(_num_workers) .set_max_threads(_num_workers) .build(&_workers)); - for (int i = 0; i < _num_workers; ++i) { - auto status = _workers->submit_func([this, i]() { - std::shared_ptr base; - { - std::lock_guard lock(_event_bases_lock); - base = _event_bases[i]; - } - /* Create a new evhttp object to handle requests. */ - std::shared_ptr http(evhttp_new(base.get()), + // Pre-create all evhttp objects and store them as class members + // to ensure proper lifecycle management during shutdown + { + std::lock_guard lock(_event_bases_lock); + _evhttp_servers.resize(_num_workers); + for (int i = 0; i < _num_workers; ++i) { + std::shared_ptr http(evhttp_new(_event_bases[i].get()), [](evhttp* http) { evhttp_free(http); }); CHECK(http != nullptr) << "Couldn't create an evhttp."; @@ -136,6 +134,17 @@ void EvHttpServer::start() { evhttp_set_newreqcb(http.get(), on_connection, this); evhttp_set_gencb(http.get(), on_request, this); + _evhttp_servers[i] = http; + } + } + + for (int i = 0; i < _num_workers; ++i) { + auto status = _workers->submit_func([this, i]() { + std::shared_ptr base; + { + std::lock_guard lock(_event_bases_lock); + base = _event_bases[i]; + } event_base_dispatch(base.get()); }); CHECK(status.ok()); @@ -143,15 +152,31 @@ void EvHttpServer::start() { } void EvHttpServer::stop() { + // 1. Close server fd first to reject new connections + close(_server_fd); + _server_fd = -1; + + // 2. Break all event loops to make dispatch return { std::lock_guard lock(_event_bases_lock); for (int i = 0; i < _num_workers; ++i) { - event_base_loopbreak(_event_bases[i].get()); + if (_event_bases[i]) { + event_base_loopbreak(_event_bases[i].get()); + } } } + + // 3. Wait for all worker threads to finish event_base_dispatch _workers->shutdown(); - _event_bases.clear(); - close(_server_fd); + + // 4. Now it's safe to cleanup - all worker threads have exited + // Clear evhttp before event_base since evhttp depends on event_base + { + std::lock_guard lock(_event_bases_lock); + _evhttp_servers.clear(); + _event_bases.clear(); + } + _started = false; } diff --git a/be/src/http/ev_http_server.h b/be/src/http/ev_http_server.h index d74a8cb4efd283..fa1c5a27928dc9 100644 --- a/be/src/http/ev_http_server.h +++ b/be/src/http/ev_http_server.h @@ -27,6 +27,7 @@ #include "util/path_trie.hpp" struct event_base; +struct evhttp; namespace doris { @@ -74,8 +75,9 @@ class EvHttpServer { int _server_fd = -1; std::unique_ptr _workers; - std::mutex _event_bases_lock; // protect _event_bases + std::mutex _event_bases_lock; // protect _event_bases and _evhttp_servers std::vector> _event_bases; + std::vector> _evhttp_servers; std::mutex _handler_lock; PathTrie _get_handlers;