Fix TopicManager shutdown deadlock with connected clients

core_sim/src/topic_manager.cpp

@@ -12,6 +12,7 @@
 #include <functional>
 #include <map>
 #include <memory>
+#include <mutex>
 #include <shared_mutex>
 #include <sstream>
 #include <stdexcept>
@@ -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<std::mutex> lifecycle_lock(lifecycle_lock_);
   std::unique_lock<std::shared_timed_mutex> 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<std::mutex> lifecycle_lock(lifecycle_lock_);
   std::unique_lock<std::shared_timed_mutex> 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);
 

core_sim/test/gtest_topic_manager.cpp

@@ -4,7 +4,11 @@
 // MIT License. All rights reserved.
 
 #include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <future>
 #include <string>
+#include <thread>
 
 #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<char*>(frame.data()),
+                                     frame.size(), 0));
+        }
+        std::promise<void> 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), "");
+}