Shutdown regressions in signal handler (from #848)

src/WaitHelpers_TEST.cc

@@ -23,6 +23,11 @@
 #include <csignal>
 #include <thread>
 
+#ifndef _WIN32
+  #include <pthread.h>
+  #include <unistd.h>
+#endif
+
 #include "gz/transport/WaitHelpers.hh"
 #include "gz/transport/Node.hh"
 #include "test_utils.hh"
@@ -31,6 +36,159 @@
 
 using namespace gz;
 
+#ifndef _WIN32
+namespace
+{
+//////////////////////////////////////////////////
+void NoopSignalHandler(int)
+{
+}
+
+//////////////////////////////////////////////////
+void VerifyWaitForShutdownBurstSignalsStayNonBlocking()
+{
+  std::thread waiter([] { transport::waitForShutdown(); });
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  raise(SIGINT);
+  waiter.join();
+
+  std::atomic<int> signalCount{0};
+  std::atomic<bool> finished{false};
+  std::thread signalThread([&]
+  {
+    for (int i = 0; i < 200000; ++i)
+    {
+      raise(SIGINT);
+      signalCount.store(i + 1);
+    }
+    finished.store(true);
+  });
+
+  int lastCount = 0;
+  int stalledChecks = 0;
+  while (!finished.load())
+  {
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    const int currentCount = signalCount.load();
+    if (currentCount == lastCount)
+    {
+      ++stalledChecks;
+      if (stalledChecks >= 10)
+        _exit(1);
+    }
+    else
+    {
+      stalledChecks = 0;
+      lastCount = currentCount;
+    }
+  }
+
+  signalThread.join();
+  _exit(0);
+}
+
+//////////////////////////////////////////////////
+void VerifyWaitForShutdownPreservesErrno()
+{
+  struct sigaction action = {};
+  action.sa_handler = NoopSignalHandler;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGUSR1, &action, nullptr);
+
+  std::thread waiter([] { transport::waitForShutdown(); });
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  raise(SIGINT);
+  waiter.join();
+
+  std::atomic<int> signalCount{0};
+  std::atomic<bool> finished{false};
+  std::atomic<bool> stopRequested{false};
+  std::atomic<int> observedErrno{0};
+  std::thread signalThread([&]
+  {
+    errno = EBUSY;
+    for (int i = 0; i < 200000; ++i)
+    {
+      raise(SIGINT);
+      signalCount.store(i + 1);
+      if (stopRequested.load())
+        break;
+    }
+    observedErrno.store(errno);
+    finished.store(true);
+  });
+
+  int lastCount = 0;
+  int stalledChecks = 0;
+  while (!finished.load())
+  {
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    const int currentCount = signalCount.load();
+    if (currentCount == lastCount)
+    {
+      ++stalledChecks;
+      if (stalledChecks >= 10)
+      {
+        stopRequested.store(true);
+        pthread_kill(signalThread.native_handle(), SIGUSR1);
+        break;
+      }
+    }
+    else
+    {
+      stalledChecks = 0;
+      lastCount = currentCount;
+    }
+  }
+
+  for (int i = 0; i < 50 && !finished.load(); ++i)
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+
+  if (finished.load())
+  {
+    signalThread.join();
+    _exit(observedErrno.load() == EBUSY ? 0 : 1);
+  }
+
+  _exit(1);
+}
+
+//////////////////////////////////////////////////
+void VerifyWaitForShutdownWakesAllWaiters()
+{
+  std::atomic<int> readyCount{0};
+  std::atomic<int> completedCount{0};
+
+  auto waiter = [&]
+  {
+    ++readyCount;
+    transport::waitForShutdown();
+    ++completedCount;
+  };
+
+  std::thread firstThread(waiter);
+  std::thread secondThread(waiter);
+  (void)firstThread;
+  (void)secondThread;
+
+  for (int i = 0; i < 50 && readyCount.load() < 2; ++i)
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  raise(SIGINT);
+  for (int i = 0; i < 50 && completedCount.load() < 2; ++i)
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+  if (completedCount.load() != 2)
+    _exit(1);
+
+  firstThread.join();
+  secondThread.join();
+  _exit(0);
+}
+}  // namespace
+#endif
+
 //////////////////////////////////////////////////
 /// \brief waitUntil: predicate immediately returns true near-instantly.
 TEST(WaitHelpersTest, WaitUntilImmediatelyTrue)
@@ -174,3 +332,46 @@ TEST(WaitHelpersTest, waitForShutdownReEntryStress)
     aThread.join();
   }
 }
+
+#ifndef _WIN32
+#if defined(__clang__)
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wswitch-default"
+#endif
+//////////////////////////////////////////////////
+/// \brief One shutdown signal should release every blocked waiter. The current
+/// implementation only wakes one waiter, so this test fails in a plain build
+/// until the process-wide wake semantics are restored.
+TEST(WaitHelpersTest, waitForShutdownSingleSignalOnlyWakesOneWaiter)
+{
+  GTEST_FLAG_SET(death_test_style, "threadsafe");
+  ASSERT_EXIT(VerifyWaitForShutdownWakesAllWaiters(),
+      ::testing::ExitedWithCode(0), "");
+}
+
+//////////////////////////////////////////////////
+/// \brief Repeated shutdown signals should not block once there is no waiter
+/// draining the persistent self-pipe. The current blocking pipe makes the
+/// signal path stall, so this test fails in a plain build until the pipe is
+/// made nonblocking.
+TEST(WaitHelpersTest, waitForShutdownBurstSignalsCanBlockSignalHandler)
+{
+  GTEST_FLAG_SET(death_test_style, "threadsafe");
+  ASSERT_EXIT(VerifyWaitForShutdownBurstSignalsStayNonBlocking(),
+      ::testing::ExitedWithCode(0), "");
+}
+
+//////////////////////////////////////////////////
+/// \brief A shutdown signal should preserve the interrupted thread's errno.
+/// The current handler leaks the write() failure errno back to the caller, so
+/// this test fails in a plain build until errno is saved and restored.
+TEST(WaitHelpersTest, waitForShutdownSignalHandlerCanClobberErrno)
+{
+  GTEST_FLAG_SET(death_test_style, "threadsafe");
+  ASSERT_EXIT(VerifyWaitForShutdownPreservesErrno(),
+      ::testing::ExitedWithCode(0), "");
+}
+#if defined(__clang__)
+#pragma clang diagnostic pop
+#endif
+#endif