MainLoop: Add unit tests

[labath]

Summary

This adds a couple of unit tests to the MainLoop class. To get the
kqueue based version of the signal handling passing, I needed to
modify the implementation a bit to make the queue object persistent.
Otherwise, only the signals which are send during the Run call would get
processed, which did not match the ppoll behaviour.

I also took the opportunity to remove the ForEach template functions and
replace them with something more reasonable.

[eugene]

source/Host/common/MainLoop.cpp

66 ↗

(On Diff #97464)

Could you please add here a comment describing when kqueue/pselect/ppoll are used.

lldb/trunk/include/lldb/Host/MainLoop.h

 // TODO: Add locking if this class is to be used in a multi-threaded context.
 class MainLoop : public MainLoopBase {
 private:
   class SignalHandle;
 
 public:
   typedef std::unique_ptr<SignalHandle> SignalHandleUP;
 
+  MainLoop();
   ~MainLoop() override;
 
   ReadHandleUP RegisterReadObject(const lldb::IOObjectSP &object_sp,
                                   const Callback &callback,
                                   Error &error) override;
 
   // Listening for signals from multiple MainLoop instances is perfectly safe as
   // long as they don't try to listen for the same signal. The callback function
public:
   void RequestTermination() override { m_terminate_request = true; }
 
 protected:
   void UnregisterReadObject(IOObject::WaitableHandle handle) override;
 
   void UnregisterSignal(int signo);
 
 private:
+  void ProcessReadObject(IOObject::WaitableHandle handle);
+  void ProcessSignal(int signo);
+
   class SignalHandle {
   public:
     ~SignalHandle() { m_mainloop.UnregisterSignal(m_signo); }
 
   private:
     SignalHandle(MainLoop &mainloop, int signo)
         : m_mainloop(mainloop), m_signo(signo) {}
 
#if HAVE_SIGACTION
     struct sigaction old_action;
 #endif
     bool was_blocked : 1;
   };
   class RunImpl;
 
   llvm::DenseMap<IOObject::WaitableHandle, Callback> m_read_fds;
   llvm::DenseMap<int, SignalInfo> m_signals;
+#if HAVE_SYS_EVENT_H
+  int m_kqueue;
+#endif
   bool m_terminate_request : 1;
 };
 
 } // namespace lldb_private
 
 #endif // lldb_Host_MainLoop_h_

lldb/trunk/source/Host/common/MainLoop.cpp

 #include "lldb/Utility/Error.h"
 #include <algorithm>
 #include <cassert>
 #include <cerrno>
 #include <csignal>
 #include <vector>
 #include <time.h>
 
+// Multiplexing is implemented using kqueue on systems that support it (BSD
+// variants including OSX). On linux we use ppoll, while android uses pselect
+// (ppoll is present but not implemented properly). On windows we use WSApoll
+// (which does not support signals).
+
 #if HAVE_SYS_EVENT_H
 #include <sys/event.h>
 #elif defined(LLVM_ON_WIN32)
 #include <winsock2.h>
 #else
 #include <poll.h>
 #endif
 

 
 static void SignalHandler(int signo, siginfo_t *info, void *) {
   assert(signo < NSIG);
   g_signal_flags[signo] = 1;
 }
 
 class MainLoop::RunImpl {
 public:
-  // TODO: Use llvm::Expected<T>
-  static std::unique_ptr<RunImpl> Create(MainLoop &loop, Error &error);
-  ~RunImpl();
+  RunImpl(MainLoop &loop);
+  ~RunImpl() = default;
 
   Error Poll();
-
-  template <typename F> void ForEachReadFD(F &&f);
-  template <typename F> void ForEachSignal(F &&f);
+  void ProcessEvents();
 
 private:
   MainLoop &loop;
 
 #if HAVE_SYS_EVENT_H
-  int queue_id;
   std::vector<struct kevent> in_events;
   struct kevent out_events[4];
   int num_events = -1;
 
-  RunImpl(MainLoop &loop, int queue_id) : loop(loop), queue_id(queue_id) {
-    in_events.reserve(loop.m_read_fds.size() + loop.m_signals.size());
-  }
 #else
-  std::vector<int> signals;
 #ifdef FORCE_PSELECT
   fd_set read_fd_set;
 #else
   std::vector<struct pollfd> read_fds;
 #endif
 
-  RunImpl(MainLoop &loop) : loop(loop) {
-    signals.reserve(loop.m_signals.size());
-  }
-
   sigset_t get_sigmask();
 #endif
 };
 
 #if HAVE_SYS_EVENT_H
-MainLoop::RunImpl::~RunImpl() {
-  int r = close(queue_id);
-  assert(r == 0);
-  (void)r;
-}
-std::unique_ptr<MainLoop::RunImpl> MainLoop::RunImpl::Create(MainLoop &loop, Error &error)
-{
-  error.Clear();
-  int queue_id = kqueue();
-  if(queue_id < 0) {
-    error = Error(errno, eErrorTypePOSIX);
-    return nullptr;
-  }
-  return std::unique_ptr<RunImpl>(new RunImpl(loop, queue_id));
+MainLoop::RunImpl::RunImpl(MainLoop &loop) : loop(loop) {
+  in_events.reserve(loop.m_read_fds.size());
 }
 
 Error MainLoop::RunImpl::Poll() {
-  in_events.resize(loop.m_read_fds.size() + loop.m_signals.size());
+  in_events.resize(loop.m_read_fds.size());
   unsigned i = 0;
   for (auto &fd : loop.m_read_fds)
     EV_SET(&in_events[i++], fd.first, EVFILT_READ, EV_ADD, 0, 0, 0);
 
-  for (const auto &sig : loop.m_signals)
-    EV_SET(&in_events[i++], sig.first, EVFILT_SIGNAL, EV_ADD, 0, 0, 0);
-
-  num_events = kevent(queue_id, in_events.data(), in_events.size(), out_events,
-                      llvm::array_lengthof(out_events), nullptr);
+  num_events = kevent(loop.m_kqueue, in_events.data(), in_events.size(),
+                      out_events, llvm::array_lengthof(out_events), nullptr);
 
   if (num_events < 0)
     return Error("kevent() failed with error %d\n", num_events);
   return Error();
 }
 
-template <typename F> void MainLoop::RunImpl::ForEachReadFD(F &&f) {
+void MainLoop::RunImpl::ProcessEvents() {
   assert(num_events >= 0);
   for (int i = 0; i < num_events; ++i) {
-    f(out_events[i].ident);
     if (loop.m_terminate_request)
       return;
+    switch (out_events[i].filter) {
+    case EVFILT_READ:
+      loop.ProcessReadObject(out_events[i].ident);
+      break;
+    case EVFILT_SIGNAL:
+      loop.ProcessSignal(out_events[i].ident);
+      break;
+    default:
+      llvm_unreachable("Unknown event");
+    }
   }
 }
-template <typename F> void MainLoop::RunImpl::ForEachSignal(F && f) {}
 #else
-MainLoop::RunImpl::~RunImpl() {}
-std::unique_ptr<MainLoop::RunImpl> MainLoop::RunImpl::Create(MainLoop &loop, Error &error)
-{
-  error.Clear();
-  return std::unique_ptr<RunImpl>(new RunImpl(loop));
+MainLoop::RunImpl::RunImpl(MainLoop &loop) : loop(loop) {
+#ifndef FORCE_PSELECT
+  read_fds.reserve(loop.m_read_fds.size());
+#endif
 }
 
 sigset_t MainLoop::RunImpl::get_sigmask() {
 #if SIGNAL_POLLING_UNSUPPORTED
   return 0;
 #else
   sigset_t sigmask;
   int ret = pthread_sigmask(SIG_SETMASK, nullptr, &sigmask);
   assert(ret == 0);
   (void) ret;
 
-  for (const auto &sig : loop.m_signals) {
-    signals.push_back(sig.first);
+  for (const auto &sig : loop.m_signals)
     sigdelset(&sigmask, sig.first);
-  }
   return sigmask;
 #endif
 }
 
 #ifdef FORCE_PSELECT
 Error MainLoop::RunImpl::Poll() {
-  signals.clear();
-
   FD_ZERO(&read_fd_set);
   int nfds = 0;
   for (const auto &fd : loop.m_read_fds) {
     FD_SET(fd.first, &read_fd_set);
     nfds = std::max(nfds, fd.first + 1);
   }
 
   sigset_t sigmask = get_sigmask();
   if (pselect(nfds, &read_fd_set, nullptr, nullptr, nullptr, &sigmask) == -1 &&
       errno != EINTR)
     return Error(errno, eErrorTypePOSIX);
 
   return Error();
 }
-
-template <typename F> void MainLoop::RunImpl::ForEachReadFD(F &&f) {
-  for (const auto &fd : loop.m_read_fds) {
-    if(!FD_ISSET(fd.first, &read_fd_set))
-      continue;
-
-    f(fd.first);
-    if (loop.m_terminate_request)
-      return;
-  }
-}
 #else
 Error MainLoop::RunImpl::Poll() {
-  signals.clear();
   read_fds.clear();
 
   sigset_t sigmask = get_sigmask();
 
   for (const auto &fd : loop.m_read_fds) {
     struct pollfd pfd;
     pfd.fd = fd.first;
     pfd.events = POLLIN;
     pfd.revents = 0;
     read_fds.push_back(pfd);
   }
 
   if (ppoll(read_fds.data(), read_fds.size(), nullptr, &sigmask) == -1 &&
       errno != EINTR)
     return Error(errno, eErrorTypePOSIX);
 
   return Error();
 }
+#endif
 
-template <typename F> void MainLoop::RunImpl::ForEachReadFD(F &&f) {
+void MainLoop::RunImpl::ProcessEvents() {
+#ifdef FORCE_PSELECT
+  for (const auto &fd : loop.m_read_fds) {
+    if (!FD_ISSET(fd.first, &read_fd_set))
+      continue;
+    IOObject::WaitableHandle handle = fd.first;
+#else
   for (const auto &fd : read_fds) {
     if ((fd.revents & POLLIN) == 0)
       continue;
-
-    f(fd.fd);
+    IOObject::WaitableHandle handle = fd.fd;
+#endif
     if (loop.m_terminate_request)
       return;
-  }
-}
-#endif
 
-template <typename F> void MainLoop::RunImpl::ForEachSignal(F &&f) {
-  for (int sig : signals) {
-    if (g_signal_flags[sig] == 0)
-      continue; // No signal
-    g_signal_flags[sig] = 0;
-    f(sig);
+    loop.ProcessReadObject(handle);
+  }
 
+  for (const auto &entry : loop.m_signals) {
     if (loop.m_terminate_request)
       return;
+    if (g_signal_flags[entry.first] == 0)
+      continue; // No signal
+    g_signal_flags[entry.first] = 0;
+    loop.ProcessSignal(entry.first);
   }
 }
 #endif
 
+MainLoop::MainLoop() {
+#if HAVE_SYS_EVENT_H
+  m_kqueue = kqueue();
+  assert(m_kqueue >= 0);
+#endif
+}
 MainLoop::~MainLoop() {
+#if HAVE_SYS_EVENT_H
+  close(m_kqueue);
+#endif
   assert(m_read_fds.size() == 0);
   assert(m_signals.size() == 0);
 }
 
 MainLoop::ReadHandleUP
 MainLoop::RegisterReadObject(const IOObjectSP &object_sp,
                                   const Callback &callback, Error &error) {
 #ifdef LLVM_ON_WIN32
#else
 
   SignalInfo info;
   info.callback = callback;
   struct sigaction new_action;
   new_action.sa_sigaction = &SignalHandler;
   new_action.sa_flags = SA_SIGINFO;
   sigemptyset(&new_action.sa_mask);
   sigaddset(&new_action.sa_mask, signo);
-
   sigset_t old_set;
-  if (int ret = pthread_sigmask(SIG_BLOCK, &new_action.sa_mask, &old_set)) {
-    error.SetErrorStringWithFormat("pthread_sigmask failed with error %d\n",
-                                   ret);
-    return nullptr;
-  }
 
-  info.was_blocked = sigismember(&old_set, signo);
-  if (sigaction(signo, &new_action, &info.old_action) == -1) {
-    error.SetErrorToErrno();
-    if (!info.was_blocked)
-      pthread_sigmask(SIG_UNBLOCK, &new_action.sa_mask, nullptr);
-    return nullptr;
-  }
+  g_signal_flags[signo] = 0;
 
+  // Even if using kqueue, the signal handler will still be invoked, so it's
+  // important to replace it with our "bening" handler.
+  int ret = sigaction(signo, &new_action, &info.old_action);
+  assert(ret == 0 && "sigaction failed");
+
+#if HAVE_SYS_EVENT_H
+  struct kevent ev;
+  EV_SET(&ev, signo, EVFILT_SIGNAL, EV_ADD, 0, 0, 0);
+  ret = kevent(m_kqueue, &ev, 1, nullptr, 0, nullptr);
+  assert(ret == 0);
+#endif
+
+  // If we're using kqueue, the signal needs to be unblocked in order to recieve
+  // it. If using pselect/ppoll, we need to block it, and later unblock it as a
+  // part of the system call.
+  ret = pthread_sigmask(HAVE_SYS_EVENT_H ? SIG_UNBLOCK : SIG_BLOCK,
+                        &new_action.sa_mask, &old_set);
+  assert(ret == 0 && "pthread_sigmask failed");
+  info.was_blocked = sigismember(&old_set, signo);
   m_signals.insert({signo, info});
-  g_signal_flags[signo] = 0;
 
   return SignalHandleUP(new SignalHandle(*this, signo));
 #endif
 }
 
 void MainLoop::UnregisterReadObject(IOObject::WaitableHandle handle) {
   bool erased = m_read_fds.erase(handle);
   UNUSED_IF_ASSERT_DISABLED(erased);
   assert(erased);
 }
 
 void MainLoop::UnregisterSignal(int signo) {
 #if SIGNAL_POLLING_UNSUPPORTED
   Error("Signal polling is not supported on this platform.");
 #else
-  // We undo the actions of RegisterSignal on a best-effort basis.
   auto it = m_signals.find(signo);
   assert(it != m_signals.end());
 
   sigaction(signo, &it->second.old_action, nullptr);
 
   sigset_t set;
   sigemptyset(&set);
   sigaddset(&set, signo);
-  pthread_sigmask(it->second.was_blocked ? SIG_BLOCK : SIG_UNBLOCK, &set,
-                  nullptr);
+  int ret = pthread_sigmask(it->second.was_blocked ? SIG_BLOCK : SIG_UNBLOCK,
+                            &set, nullptr);
+  assert(ret == 0);
+  (void)ret;
+
+#if HAVE_SYS_EVENT_H
+  struct kevent ev;
+  EV_SET(&ev, signo, EVFILT_SIGNAL, EV_DELETE, 0, 0, 0);
+  ret = kevent(m_kqueue, &ev, 1, nullptr, 0, nullptr);
+  assert(ret == 0);
+#endif
 
   m_signals.erase(it);
 #endif
 }
 
 Error MainLoop::Run() {
   m_terminate_request = false;
   
   Error error;
-  auto impl = RunImpl::Create(*this, error);
-  if (!impl)
-    return error;
+  RunImpl impl(*this);
 
   // run until termination or until we run out of things to listen to
   while (!m_terminate_request && (!m_read_fds.empty() || !m_signals.empty())) {
 
-    error = impl->Poll();
+    error = impl.Poll();
     if (error.Fail())
       return error;
 
-    impl->ForEachSignal([&](int sig) {
-      auto it = m_signals.find(sig);
-      if (it != m_signals.end())
-        it->second.callback(*this); // Do the work
-    });
-    if (m_terminate_request)
-      return Error();
+    impl.ProcessEvents();
 
-    impl->ForEachReadFD([&](int fd) {
-      auto it = m_read_fds.find(fd);
-      if (it != m_read_fds.end())
-        it->second(*this); // Do the work
-    });
     if (m_terminate_request)
       return Error();
   }
   return Error();
 }
+
+void MainLoop::ProcessSignal(int signo) {
+  auto it = m_signals.find(signo);
+  if (it != m_signals.end())
+    it->second.callback(*this); // Do the work
+}
+
+void MainLoop::ProcessReadObject(IOObject::WaitableHandle handle) {
+  auto it = m_read_fds.find(handle);
+  if (it != m_read_fds.end())
+    it->second(*this); // Do the work
+}

lldb/trunk/unittests/Host/CMakeLists.txt

 set (FILES
   FileSpecTest.cpp
   FileSystemTest.cpp
+  MainLoopTest.cpp
   SocketAddressTest.cpp
   SocketTest.cpp
   SymbolsTest.cpp
 )
 
 if (CMAKE_SYSTEM_NAME MATCHES "Linux|Android")
   list(APPEND FILES
     linux/HostTest.cpp

lldb/trunk/unittests/Host/MainLoopTest.cpp

+//===-- MainLoopTest.cpp ----------------------------------------*- C++ -*-===//
+//
+//                     The LLVM Compiler Infrastructure
+//
+// This file is distributed under the University of Illinois Open Source
+// License. See LICENSE.TXT for details.
+//
+//===----------------------------------------------------------------------===//
+
+#include "lldb/Host/MainLoop.h"
+#include "lldb/Host/common/TCPSocket.h"
+#include "gtest/gtest.h"
+#include <future>
+
+using namespace lldb_private;
+
+namespace {
+class MainLoopTest : public testing::Test {
+public:
+  static void SetUpTestCase() {
+#ifdef _MSC_VER
+    WSADATA data;
+    ASSERT_EQ(0, WSAStartup(MAKEWORD(2, 2), &data));
+#endif
+  }
+
+  static void TearDownTestCase() {
+#ifdef _MSC_VER
+    ASSERT_EQ(0, WSACleanup());
+#endif
+  }
+
+  void SetUp() override {
+    bool child_processes_inherit = false;
+    Error error;
+    std::unique_ptr<TCPSocket> listen_socket_up(
+        new TCPSocket(true, child_processes_inherit));
+    ASSERT_TRUE(error.Success());
+    error = listen_socket_up->Listen("localhost:0", 5);
+    ASSERT_TRUE(error.Success());
+
+    Socket *accept_socket;
+    std::future<Error> accept_error = std::async(std::launch::async, [&] {
+      return listen_socket_up->Accept(accept_socket);
+    });
+
+    std::unique_ptr<TCPSocket> connect_socket_up(
+        new TCPSocket(true, child_processes_inherit));
+    error = connect_socket_up->Connect(
+        llvm::formatv("localhost:{0}", listen_socket_up->GetLocalPortNumber())
+            .str());
+    ASSERT_TRUE(error.Success());
+    ASSERT_TRUE(accept_error.get().Success());
+
+    callback_count = 0;
+    socketpair[0] = std::move(connect_socket_up);
+    socketpair[1].reset(accept_socket);
+  }
+
+  void TearDown() override {
+    socketpair[0].reset();
+    socketpair[1].reset();
+  }
+
+protected:
+  MainLoop::Callback make_callback() {
+    return [&](MainLoopBase &loop) {
+      ++callback_count;
+      loop.RequestTermination();
+    };
+  }
+  std::shared_ptr<Socket> socketpair[2];
+  unsigned callback_count;
+};
+} // namespace
+
+TEST_F(MainLoopTest, ReadObject) {
+  char X = 'X';
+  size_t len = sizeof(X);
+  ASSERT_TRUE(socketpair[0]->Write(&X, len).Success());
+
+  MainLoop loop;
+
+  Error error;
+  auto handle = loop.RegisterReadObject(socketpair[1], make_callback(), error);
+  ASSERT_TRUE(error.Success());
+  ASSERT_TRUE(handle);
+  ASSERT_TRUE(loop.Run().Success());
+  ASSERT_EQ(1u, callback_count);
+}
+
+TEST_F(MainLoopTest, TerminatesImmediately) {
+  char X = 'X';
+  size_t len = sizeof(X);
+  ASSERT_TRUE(socketpair[0]->Write(&X, len).Success());
+  ASSERT_TRUE(socketpair[1]->Write(&X, len).Success());
+
+  MainLoop loop;
+  Error error;
+  auto handle0 = loop.RegisterReadObject(socketpair[0], make_callback(), error);
+  ASSERT_TRUE(error.Success());
+  auto handle1 = loop.RegisterReadObject(socketpair[1], make_callback(), error);
+  ASSERT_TRUE(error.Success());
+
+  ASSERT_TRUE(loop.Run().Success());
+  ASSERT_EQ(1u, callback_count);
+}
+
+#ifdef LLVM_ON_UNIX
+TEST_F(MainLoopTest, Signal) {
+  MainLoop loop;
+  Error error;
+
+  auto handle = loop.RegisterSignal(SIGUSR1, make_callback(), error);
+  ASSERT_TRUE(error.Success());
+  kill(getpid(), SIGUSR1);
+  ASSERT_TRUE(loop.Run().Success());
+  ASSERT_EQ(1u, callback_count);
+}
+#endif