diff --git a/src/mtconnect/configuration/async_context.hpp b/src/mtconnect/configuration/async_context.hpp index 910c07ee1..8723df47c 100644 --- a/src/mtconnect/configuration/async_context.hpp +++ b/src/mtconnect/configuration/async_context.hpp @@ -71,7 +71,24 @@ namespace mtconnect::configuration { { for (int i = 0; i < m_threadCount; i++) { - m_workers.emplace_back(boost::thread([this]() { m_context.run(); })); + m_workers.emplace_back(boost::thread([this]() { + while (m_running) + { + try + { + m_context.run(); + break; // clean return: context stopped or out of work + } + catch (const std::exception &e) + { + LOG(error) << "Unhandled exception in io_context worker, resuming: " << e.what(); + } + catch (...) + { + LOG(error) << "Unhandled unknown exception in io_context worker, resuming"; + } + } + })); } auto &first = m_workers.front(); while (m_running && !m_paused) diff --git a/src/mtconnect/sink/rest_sink/server.cpp b/src/mtconnect/sink/rest_sink/server.cpp index dd34add95..87530d902 100644 --- a/src/mtconnect/sink/rest_sink/server.cpp +++ b/src/mtconnect/sink/rest_sink/server.cpp @@ -49,6 +49,10 @@ namespace mtconnect::sink::rest_sink { using std::placeholders::_1; using std::placeholders::_2; + // Delay before re-arming the accept loop after a recoverable accept error so we + // don't hot-loop while a transient condition (e.g. fd exhaustion) clears. + static constexpr auto ACCEPT_BACKOFF = std::chrono::milliseconds(100); + void Server::loadTlsCertificate() { if (HasOption(m_options, configuration::TlsCertificateChain) && @@ -117,19 +121,19 @@ namespace mtconnect::sink::rest_sink { if (ec) { fail(ec, "Cannot open server socket"); - return; + throw std::runtime_error(std::string("Cannot open server socket: ") + ec.message()); } m_acceptor.set_option(boost::asio::socket_base::reuse_address(true), ec); if (ec) { fail(ec, "Cannot set reuse address"); - return; + throw std::runtime_error(std::string("Cannot set reuse address: ") + ec.message()); } m_acceptor.bind(ep, ec); if (ec) { fail(ec, "Cannot bind to server address"); - return; + throw std::runtime_error(std::string("Cannot bind to server address: ") + ec.message()); } if (m_port == 0) { @@ -140,7 +144,7 @@ namespace mtconnect::sink::rest_sink { if (ec) { fail(ec, "Cannot set listen queue length"); - return; + throw std::runtime_error(std::string("Cannot set listen queue length: ") + ec.message()); } m_listening = true; @@ -177,13 +181,39 @@ namespace mtconnect::sink::rest_sink { { NAMED_SCOPE("Server::accept"); + if (ec == boost::asio::error::operation_aborted) + { + // The acceptor was closed during shutdown, do not re-arm. + return; + } + if (ec) { LOG(error) << ec.category().message(ec.value()) << ": " << ec.message(); fail(ec, "Accept failed"); + + // The acceptor remains open and valid after a non-fatal accept error, so + // re-arm after a short backoff to avoid hot-looping while a transient + // condition (e.g. fd exhaustion) clears. + if (m_run) + { + m_acceptBackoff.expires_after(ACCEPT_BACKOFF); + m_acceptBackoff.async_wait([this](beast::error_code tec) { + if (!tec && m_run) + { + // Count only error-recovery re-arms (not the success-path re-arm or + // the operation_aborted shutdown path) for diagnostics / tests. + ++m_acceptErrorRearms; + m_acceptor.async_accept(net::make_strand(m_context), + beast::bind_front_handler(&Server::accept, this)); + } + }); + } + return; } - else if (m_run) + + if (m_run) { auto dispatcher = [this](SessionPtr session, RequestPtr request) { if (!m_run) @@ -226,7 +256,7 @@ namespace mtconnect::sink::rest_sink { // Report a failure void Server::fail(beast::error_code ec, char const *what) { - LOG(error) << " error: " << ec.message(); + LOG(error) << what << " error: " << ec.message(); } using namespace mtconnect::printer; diff --git a/src/mtconnect/sink/rest_sink/server.hpp b/src/mtconnect/sink/rest_sink/server.hpp index f025fa2b1..52b3fe796 100644 --- a/src/mtconnect/sink/rest_sink/server.hpp +++ b/src/mtconnect/sink/rest_sink/server.hpp @@ -20,10 +20,13 @@ #include #include #include +#include #include #include +#include #include +#include #include #include #include @@ -60,6 +63,7 @@ namespace mtconnect::sink::rest_sink { m_options(options), m_allowPuts(IsOptionSet(options, configuration::AllowPut)), m_acceptor(context), + m_acceptBackoff(context), m_sslContext(boost::asio::ssl::context::tls) { auto inter = GetOption(options, configuration::ServerIp); @@ -272,6 +276,10 @@ namespace mtconnect::sink::rest_sink { /// @return the error function ErrorFunction getErrorFunction() const { return m_errorFunction; } + /// @brief Number of times the accept loop re-armed after a recoverable accept error. + /// @return the cumulative re-arm count (useful for diagnostics / tests) + std::uint64_t getAcceptRearmCount() const { return m_acceptErrorRearms; } + /// @name Only for testing ///@{ /// @brief Callback for testing. Allows test to grab the last session dispatched. @@ -318,6 +326,11 @@ namespace mtconnect::sink::rest_sink { std::optional m_parameterDocumentation; boost::asio::ip::tcp::acceptor m_acceptor; + boost::asio::steady_timer m_acceptBackoff; + /// Cumulative count of accept-loop re-arms triggered by a recoverable accept + /// error (incremented only on the error-recovery path, not normal success or + /// shutdown). Exposed via getAcceptRearmCount() for diagnostics / tests. + std::atomic m_acceptErrorRearms {0}; boost::asio::ssl::context m_sslContext; bool m_tlsEnabled {false}; bool m_tlsOnly {false}; diff --git a/test_package/CMakeLists.txt b/test_package/CMakeLists.txt index 05f82d637..bc44911a4 100644 --- a/test_package/CMakeLists.txt +++ b/test_package/CMakeLists.txt @@ -289,6 +289,7 @@ add_agent_test(globals FALSE core) add_agent_test(config_parser FALSE configuration) add_agent_test(config FALSE configuration) +add_agent_test(async_context FALSE configuration) add_agent_test(change_observer FALSE observation) add_agent_test(observation TRUE observation) diff --git a/test_package/async_context_test.cpp b/test_package/async_context_test.cpp new file mode 100644 index 000000000..5e45970e0 --- /dev/null +++ b/test_package/async_context_test.cpp @@ -0,0 +1,70 @@ +// +// Copyright Copyright 2009-2024, AMT – The Association For Manufacturing Technology (“AMT”) +// All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +// Ensure that gtest is the first header otherwise Windows raises an error +#include +// Keep this comment to keep gtest.h above. (clang-format off/on is not working here!) + +#include + +#include +#include +#include +#include + +#include "mtconnect/configuration/async_context.hpp" + +using namespace std; +using namespace mtconnect; +using namespace mtconnect::configuration; + +// main +int main(int argc, char* argv[]) +{ + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} + +class AsyncContextTest : public testing::Test +{ +}; + +TEST_F(AsyncContextTest, worker_survives_a_thrown_handler) +{ + // Fix #3: a handler that throws must not kill the io_context worker. With a + // single worker thread handlers run FIFO on that one thread; without the fix + // the throwing handler tears down the worker and the second handler never + // runs. With the fix the worker catches, resumes run(), and processes it. + AsyncContext ctx; + + boost::asio::post(ctx.get(), [] { throw std::runtime_error("boom"); }); + + std::atomic secondRan {false}; + boost::asio::post(ctx.get(), [&] { secondRan = true; }); + + // start() blocks in its own run loop, so drive it on a separate thread. + std::thread t([&] { ctx.start(); }); + + // Poll up to ~2s so the test cannot hang even if the fix regresses. + for (int i = 0; i < 200 && !secondRan; ++i) + std::this_thread::sleep_for(10ms); + + ctx.stop(); + t.join(); + + EXPECT_TRUE(secondRan); +} diff --git a/test_package/http_server_test.cpp b/test_package/http_server_test.cpp index fa2607c58..fb17ec2d7 100644 --- a/test_package/http_server_test.cpp +++ b/test_package/http_server_test.cpp @@ -678,6 +678,58 @@ TEST_F(HttpServerTest, additional_header_fields) ASSERT_EQ("https://foo.example", f2->second); } +TEST_F(HttpServerTest, listen_throws_when_port_is_already_bound) +{ + // The fixture's server binds a random free port; capture the concrete port. + start(); + auto port = m_server->getPort(); + + // A second server bound to the same concrete ip:port must fail to bind. + // SO_REUSEADDR is set, but two sockets concurrently listening on the same + // 127.0.0.1:port still yield EADDRINUSE on Linux, so this is deterministic. + // + // We call listen() directly rather than start(): start() catches the throw + // and calls std::exit(1), which would terminate the test runner. + asio::io_context otherContext; + Server other(otherContext, + ConfigOptions {{configuration::Port, static_cast(port)}, + {configuration::ServerIp, "127.0.0.1"s}}); + ASSERT_THROW(other.listen(), std::runtime_error); +} + +TEST_F(HttpServerTest, accept_loop_rearms_after_recoverable_error) +{ + // Fix #1: on a recoverable accept error the accept loop must re-arm after a + // backoff instead of stopping permanently. getAcceptRearmCount() is a + // production seam that counts ONLY the error-recovery re-arm path, so a + // 0 -> 1 transition proves the fix actually re-armed the acceptor after the + // injected error (rather than the initial accept armed by start() simply + // serving a later connection, which would not distinguish fixed from broken). + // + // Injecting the error while the initial accept is still armed leaves two + // outstanding async_accept operations on the acceptor briefly. This is a + // supported Boost.Asio pattern (an acceptor may hold multiple concurrent + // pending accepts); both are cancelled cleanly (operation_aborted) when the + // server is destroyed in TearDown. + start(); + ASSERT_EQ(0u, m_server->getAcceptRearmCount()); + + // Drive the recoverable-error branch directly with a fresh (unopened) socket; + // the error branch returns before touching the socket. + m_server->accept(boost::asio::error::connection_aborted, tcp::socket(m_context)); + + // Pump the io_context until the 100ms backoff timer fires and re-arms, with a + // bounded loop so a regression (no re-arm) fails fast instead of hanging. + for (int i = 0; i < 100 && m_server->getAcceptRearmCount() == 0; ++i) + m_context.run_for(20ms); + + EXPECT_EQ(1u, m_server->getAcceptRearmCount()); + + // The server remains running and listening after recovering from the error. + EXPECT_TRUE(m_server->isRunning()); + EXPECT_TRUE(m_server->isListening()); +} + const string CertFile(TEST_RESOURCE_DIR "/user.crt"); const string KeyFile {TEST_RESOURCE_DIR "/user.key"}; const string DhFile {TEST_RESOURCE_DIR "/dh2048.pem"};