From ec60113e09d9fc985b1c35fe1c579ef73c067d91 Mon Sep 17 00:00:00 2001 From: selsta Date: Sun, 31 May 2026 02:37:15 +0200 Subject: [PATCH 1/2] p2p: make stop signal idempotent --- contrib/epee/include/net/abstract_tcp_server2.inl | 6 +++++- src/p2p/net_node.h | 1 + src/p2p/net_node.inl | 5 +++++ 3 files changed, 11 insertions(+), 1 deletion(-) diff --git a/contrib/epee/include/net/abstract_tcp_server2.inl b/contrib/epee/include/net/abstract_tcp_server2.inl index 56d3d7de2..f15928f45 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.inl +++ b/contrib/epee/include/net/abstract_tcp_server2.inl @@ -1579,7 +1579,11 @@ namespace net_utils template void boosted_tcp_server::send_stop_signal(std::function close_all_connections) { - m_stop_signal_sent = true; + if (m_stop_signal_sent.exchange(true)) + { + MDEBUG("Stop signal already sent"); + return; + } typename connection::shared_state *state = static_cast::shared_state*>(m_state.get()); state->stop_signal_sent = true; TRY_ENTRY(); diff --git a/src/p2p/net_node.h b/src/p2p/net_node.h index 8e3312f29..d7cbaa567 100644 --- a/src/p2p/net_node.h +++ b/src/p2p/net_node.h @@ -451,6 +451,7 @@ namespace nodetool bool m_use_ipv6; bool m_require_ipv4; std::atomic is_closing; + std::atomic m_stop_signal_sent_once{false}; std::unique_ptr mPeersLoggerThread; //critical_section m_connections_lock; //connections_indexed_container m_connections; diff --git a/src/p2p/net_node.inl b/src/p2p/net_node.inl index d1f7e7cdd..4fbc36031 100644 --- a/src/p2p/net_node.inl +++ b/src/p2p/net_node.inl @@ -1107,6 +1107,11 @@ namespace nodetool template bool node_server::send_stop_signal() { + if (m_stop_signal_sent_once.exchange(true)) + { + MDEBUG("[node] Stop signal already sent"); + return true; + } MDEBUG("[node] stopping server payload handler"); m_payload_handler.stop(); MDEBUG("[node] sending stop signal"); From b7ca9e73f23cfb0ce1e795b2074ad12af56b04e3 Mon Sep 17 00:00:00 2001 From: selsta Date: Sun, 31 May 2026 03:09:17 +0200 Subject: [PATCH 2/2] p2p: close zone connections before stopping net servers --- .../epee/include/net/abstract_tcp_server2.h | 17 +++++- .../epee/include/net/abstract_tcp_server2.inl | 57 +++++++++++++------ src/p2p/net_node.inl | 45 +++++++++------ tests/unit_tests/epee_boosted_tcp_server.cpp | 11 ++-- 4 files changed, 88 insertions(+), 42 deletions(-) diff --git a/contrib/epee/include/net/abstract_tcp_server2.h b/contrib/epee/include/net/abstract_tcp_server2.h index fe201f155..75a26cbbb 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.h +++ b/contrib/epee/include/net/abstract_tcp_server2.h @@ -320,7 +320,7 @@ namespace net_utils bool speed_limit_is_enabled() const; ///< tells us should we be sleeping here (e.g. do not sleep on RPC connections) - bool cancel(); + bool cancel(bool wait_for_shutdown = false); private: //----------------- i_service_endpoint --------------------- @@ -378,8 +378,21 @@ namespace net_utils /// wait for service workers stop bool timed_wait_server_stop(uint64_t wait_mseconds); + /// Mark the server as stopping without closing connections or stopping the io_context. + bool mark_stop_signal_sent(); + + /// Close boosted_tcp_server-owned connections, including ones not yet registered with the protocol handler. + void close_server_connections(); + + /// Stop the server io_context. + void stop_io_context(); + /// Stop the server. - void send_stop_signal(std::function close_all_connections = [](){}); + /// + /// Warning: Do NOT call this if the io_context is shared for connections + /// managed outside the boosted_tcp_server. See p2p net_node shutdown for + /// the correct staged shutdown in that case. + void send_stop_signal(); bool is_stop_signal_sent() const noexcept { return m_stop_signal_sent; }; diff --git a/contrib/epee/include/net/abstract_tcp_server2.inl b/contrib/epee/include/net/abstract_tcp_server2.inl index f15928f45..603a92841 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.inl +++ b/contrib/epee/include/net/abstract_tcp_server2.inl @@ -1119,9 +1119,9 @@ namespace net_utils } template - bool connection::cancel() + bool connection::cancel(const bool wait_for_shutdown) { - return close(false); + return close(wait_for_shutdown); } template @@ -1286,7 +1286,7 @@ namespace net_utils template boosted_tcp_server::~boosted_tcp_server() { - this->send_stop_signal(); + send_stop_signal(); timed_wait_server_stop(10000); } //--------------------------------------------------------------------------------- @@ -1577,30 +1577,55 @@ namespace net_utils } //--------------------------------------------------------------------------------- template - void boosted_tcp_server::send_stop_signal(std::function close_all_connections) + bool boosted_tcp_server::mark_stop_signal_sent() { if (m_stop_signal_sent.exchange(true)) { MDEBUG("Stop signal already sent"); - return; + return false; } typename connection::shared_state *state = static_cast::shared_state*>(m_state.get()); state->stop_signal_sent = true; - TRY_ENTRY(); - connections_mutex.lock(); - for (auto &c: connections_) + return true; + } + //--------------------------------------------------------------------------------- + template + void boosted_tcp_server::close_server_connections() + { + decltype(connections_) connections; { - c->cancel(); + boost::unique_lock lock(connections_mutex); + connections.swap(connections_); } - connections_.clear(); - connections_mutex.unlock(); - // Since we shut down connections in the strand, we want to make sure to complete the shutdown sequence before - // stopping the io_context. We let the caller handle closing because the caller is the one keeping track of all - // connections (connections_ is only a subset of all connections). - close_all_connections(); + for (auto &c: connections) + { + c->cancel(true/*wait_for_shutdown*/); + } + } + //--------------------------------------------------------------------------------- + template + void boosted_tcp_server::stop_io_context() + { + { + boost::unique_lock lock(connections_mutex); + if (!connections_.empty()) + { + MERROR("Stopping io_context with " << connections_.size() << " server-owned connections still open"); + } + } + MDEBUG("Stopping io_context"); io_context_.stop(); - MDEBUG("Done with send_stop_signal"); + } + //--------------------------------------------------------------------------------- + template + void boosted_tcp_server::send_stop_signal() + { + TRY_ENTRY(); + if (!mark_stop_signal_sent()) + return; + close_server_connections(); + stop_io_context(); CATCH_ENTRY_L0("boosted_tcp_server::send_stop_signal()", void()); } //--------------------------------------------------------------------------------- diff --git a/src/p2p/net_node.inl b/src/p2p/net_node.inl index 4fbc36031..91a54f43e 100644 --- a/src/p2p/net_node.inl +++ b/src/p2p/net_node.inl @@ -1114,26 +1114,37 @@ namespace nodetool } MDEBUG("[node] stopping server payload handler"); m_payload_handler.stop(); - MDEBUG("[node] sending stop signal"); + + MDEBUG("[node] marking net servers as stopping"); for (auto& zone : m_network_zones) { - const auto close_all_connections = [&, this]() - { - std::list connection_ids; - zone.second.m_net_server.get_config_object().foreach_connection([&](const p2p_connection_context& cntxt) { - connection_ids.push_back(cntxt.m_connection_id); - return true; - }); - for (const auto &connection_id: connection_ids) - { - MDEBUG("Closing connection " << connection_id); - // We need to wait for every connection's shutdown sequence to complete before stopping the io_context. - zone.second.m_net_server.get_config_object().close(connection_id, true/*wait_for_shutdown*/); - MDEBUG("Closed connection " << connection_id); - } - }; + zone.second.m_net_server.mark_stop_signal_sent(); + } - zone.second.m_net_server.send_stop_signal(close_all_connections); + MDEBUG("[node] closing connections"); + for (auto& zone : m_network_zones) + { + zone.second.m_net_server.close_server_connections(); + + std::list connection_ids; + zone.second.m_net_server.get_config_object().foreach_connection([&](const p2p_connection_context& cntxt) + { + connection_ids.push_back(cntxt.m_connection_id); + return true; + }); + for (const auto &connection_id: connection_ids) + { + MDEBUG("Closing connection " << connection_id); + // All zone connections must finish shutting down before any shared io_context is stopped. + zone.second.m_net_server.get_config_object().close(connection_id, true/*wait_for_shutdown*/); + MDEBUG("Closed connection " << connection_id); + } + } + + MDEBUG("[node] stopping net server io_contexts"); + for (auto& zone : m_network_zones) + { + zone.second.m_net_server.stop_io_context(); } MDEBUG("[node] Stop signal sent"); return true; diff --git a/tests/unit_tests/epee_boosted_tcp_server.cpp b/tests/unit_tests/epee_boosted_tcp_server.cpp index 6052f7a50..586cc25a5 100644 --- a/tests/unit_tests/epee_boosted_tcp_server.cpp +++ b/tests/unit_tests/epee_boosted_tcp_server.cpp @@ -801,14 +801,11 @@ TEST(boosted_tcp_server, shutdown) server.get_config_object().handshake_received.wait(); } - // Now stop the server, providing the callback necessary to wait for all connections to shutdown - const auto close_all_connections = [&]() - { - server.get_config_object().close(context.m_connection_id, true/*wait_for_shutdown*/); - }; - MINFO("Stopping the server"); - server.send_stop_signal(close_all_connections); + server.mark_stop_signal_sent(); + server.close_server_connections(); + server.get_config_object().close(context.m_connection_id, true/*wait_for_shutdown*/); + server.stop_io_context(); running_server.join(); MINFO("Waiting for handshake to cancel");