diff --git a/core_sim/src/topic_manager.cpp b/core_sim/src/topic_manager.cpp index fb37d3f1..ad7ea5d4 100644 --- a/core_sim/src/topic_manager.cpp +++ b/core_sim/src/topic_manager.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -223,6 +224,8 @@ class TopicManager::Impl { std::string local_address_; Logger log_; mutable std::shared_timed_mutex manager_lock_; + // Serialize complete Start/Stop transitions without blocking topic callbacks. + std::mutex lifecycle_lock_; std::string name_; int port_; char* recv_buffer_; @@ -379,7 +382,9 @@ void TopicManager::Impl::HandleNNGPipeEvent(nng_pipe pipe, nng_pipe_ev ev) { } void TopicManager::Impl::Start() { + std::lock_guard lifecycle_lock(lifecycle_lock_); std::unique_lock exclusive_lock(manager_lock_); + if (state_.load()) return; int rv = nng_pair0_open(&topic_socket_); if (rv != 0) { @@ -469,13 +474,21 @@ void TopicManager::Impl::Start() { } void TopicManager::Impl::Stop() { + std::lock_guard lifecycle_lock(lifecycle_lock_); std::unique_lock exclusive_lock(manager_lock_); + if (!state_.load()) return; send_dispatcher_.stop(); recv_dispatcher_.stop(); state_ = false; + // RecvLoop and the NNG disconnect callback both acquire manager_lock_. + // Waiting for either while holding it would prevent shutdown from finishing. + // lifecycle_lock_ keeps another Start/Stop from changing the socket or thread + // until this complete shutdown has finished. + exclusive_lock.unlock(); + if (recv_thread_.joinable()) { try { recv_thread_.join(); @@ -494,6 +507,7 @@ void TopicManager::Impl::Stop() { log_.LogError(name_, "nng_close failed with '%s'.", errno_str); } + exclusive_lock.lock(); topic_socket_ = NNG_SOCKET_INITIALIZER; active_pipe_count_.store(0, std::memory_order_release); diff --git a/core_sim/test/gtest_topic_manager.cpp b/core_sim/test/gtest_topic_manager.cpp index 38b8df30..a704e835 100644 --- a/core_sim/test/gtest_topic_manager.cpp +++ b/core_sim/test/gtest_topic_manager.cpp @@ -4,7 +4,11 @@ // MIT License. All rights reserved. #include +#include +#include +#include #include +#include #include "gtest/gtest.h" #include "msgpack.hpp" @@ -159,3 +163,76 @@ TEST(TopicManager, BlockingReceiveAllowsBoundedStopWithoutTraffic) { EXPECT_LT(elapsed, std::chrono::seconds(1)); } + +namespace { + +void RequireNngSuccess(int result) { + if (result != 0) { + std::fprintf(stderr, "NNG setup failed: %s\n", nng_strerror(result)); + std::_Exit(2); + } +} + +void CheckConnectedShutdown(bool pending_frames) { + // A deadlock must fail in a bounded child process, not hang the test runner. + std::thread([] { + std::this_thread::sleep_for(std::chrono::seconds(10)); + std::fprintf(stderr, "TopicManager shutdown timed out\n"); + std::_Exit(3); + }).detach(); + + projectairsim::Logger logger( + [](const std::string&, projectairsim::LogLevel, const std::string&) {}); + { + projectairsim::TopicManager manager(logger); + const int port = pending_frames ? 18992 : 18991; + manager.Load(nlohmann::json{{"ip", "127.0.0.1"}, {"port", port}}); + const std::string url = "tcp://127.0.0.1:" + std::to_string(port); + + for (int iteration = 0; iteration < 3; ++iteration) { + manager.Start(); + nng_socket peer = NNG_SOCKET_INITIALIZER; + RequireNngSuccess(nng_pair0_open(&peer)); + RequireNngSuccess(nng_socket_set_ms(peer, NNG_OPT_SENDTIMEO, 1000)); + RequireNngSuccess(nng_dial(peer, url.c_str(), nullptr, 0)); + + if (pending_frames) { + // Even an unknown topic takes manager_lock_ in RecvLoop. Queue frames + // while stopping to exercise the receive path as well as disconnect. + const std::string frame = PackTopicFrame( + projectairsim::FrameType::kSubscribe, "/shutdown-test", ""); + for (int frame_index = 0; frame_index < 16; ++frame_index) { + RequireNngSuccess(nng_send(peer, const_cast(frame.data()), + frame.size(), 0)); + } + std::promise start; + const auto ready = start.get_future().share(); + std::thread first([&] { ready.wait(); manager.Stop(); }); + std::thread second([&] { ready.wait(); manager.Stop(); }); + start.set_value(); + first.join(); + second.join(); + } else { + manager.Stop(); + } + + // Keep the peer open until Stop returns: closing it beforehand hides + // the disconnect callback's lock cycle. Repeated Stop must be harmless. + manager.Stop(); + RequireNngSuccess(nng_close(peer)); + } + } + std::_Exit(0); +} + +} // namespace + +TEST(TopicManagerDeathTest, ConnectedClientCanStopAndRestart) { + ::testing::GTEST_FLAG(death_test_style) = "threadsafe"; + ASSERT_EXIT(CheckConnectedShutdown(false), ::testing::ExitedWithCode(0), ""); +} + +TEST(TopicManagerDeathTest, PendingFramesAndConcurrentStopsCanRestart) { + ::testing::GTEST_FLAG(death_test_style) = "threadsafe"; + ASSERT_EXIT(CheckConnectedShutdown(true), ::testing::ExitedWithCode(0), ""); +}