Resurrect pselect MainLoop implementation

[labath]

Summary

It turns out that even though ppoll is available on all the android
devices we support, it does not seem to be working properly on all of
them -- MainLoop just does a busy loop with ppoll returning EINTR and
not making any progress.

This brings back the pselect implementation and makes it available on
android. I could not do any cmake checks for this as the ppoll symbol is
actually avaiable -- it just does not work.

I haven't tested (or even built) this yet on platforms other than linux,
but I wanted to share this early to get feedback on the way I abstract
the pselect/ppoll/kevent implementations.

[beanz]

This looks like a nice improvement. There are some formatting inconsistencies, can you run clang-format?

[eugene]

source/Host/common/MainLoop.cpp

82

here and below struct seems to be redundant

[zturner]

source/Host/common/MainLoop.cpp

135

Why do we need this function? Just have a function that returns an ArrayRef<int> and let the caller iterate over it themselves.

143

Same here.

167

How about just moving this #ifdef up outside of the Poll function? Given that it occurs so many times, i think it makes the logic and the differences between the two implementations easier to follow.

[labath]

source/Host/common/MainLoop.cpp

82

One of them holds the input events, and the other the output -- I am assuming they can't be the same. @beanz?

I am going to rename them to reflect this better though.

135

The logic for detecting whether an event has actually happened is different for ppoll and kevent based implementations. In case of kevent you already get a list of events ready, so it may seem that it is redundant, but in case of ppoll, I have to look through the list of input events, and check that they have they POLLIN flag set. This is basically what is this trying to abstract. I can believe it is overkill though.

I'll see if I can come up with something cleaner.

167

OK.

[labath]

Address review feedback.

[labath]

I am going to check this in to unbork the android bots. I'm happy to address any additional feedback in a followup.

I'm also planning to come some unit tests for this class next week.

include/lldb/Host/MainLoop.h

private:
 
   struct SignalInfo {
     Callback callback;
 #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;
   bool m_terminate_request : 1;
 };
 
 } // namespace lldb_private
 
 #endif // lldb_Host_MainLoop_h_

source/Host/common/MainLoop.cpp

 #endif
 
 #ifdef LLVM_ON_WIN32
 #define POLL WSAPoll
 #else
 #define POLL poll
 #endif
 
+#ifdef __ANDROID__
+#define FORCE_PSELECT
+#endif
+
 #if SIGNAL_POLLING_UNSUPPORTED
 #ifdef LLVM_ON_WIN32
 typedef int sigset_t;
 typedef int siginfo_t;
 #endif
 
 int ppoll(struct pollfd *fds, size_t nfds, const struct timespec *timeout_ts,
           const sigset_t *) {

 
 static sig_atomic_t g_signal_flags[NSIG];
 
 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();
+
+  Error Poll();
+
+  template <typename F> void ForEachReadFD(F &&f);
+  template <typename F> void ForEachSignal(F &&f);
+
+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));
+}
+
+Error MainLoop::RunImpl::Poll() {
+  in_events.resize(loop.m_read_fds.size() + loop.m_signals.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);
+
+  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) {
+  assert(num_events >= 0);
+  for (int i = 0; i < num_events; ++i) {
+    f(out_events[i].ident);
+    if (loop.m_terminate_request)
+      return;
+  }
+}
+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));
+}
+
+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);
+    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();
+}
+
+template <typename F> void MainLoop::RunImpl::ForEachReadFD(F &&f) {
+  for (const auto &fd : read_fds) {
+    if ((fd.revents & POLLIN) == 0)
+      continue;
+
+    f(fd.fd);
+    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);
+
+    if (loop.m_terminate_request)
+      return;
+  }
+}
+#endif
+
 MainLoop::~MainLoop() {
   assert(m_read_fds.size() == 0);
   assert(m_signals.size() == 0);
 }
 
 MainLoop::ReadHandleUP
 MainLoop::RegisterReadObject(const IOObjectSP &object_sp,
                                   const Callback &callback, Error &error) {
#else
   pthread_sigmask(it->second.was_blocked ? SIG_BLOCK : SIG_UNBLOCK, &set,
                   nullptr);
 
   m_signals.erase(it);
 #endif
 }
 
 Error MainLoop::Run() {
-  std::vector<int> signals;
   m_terminate_request = false;
-  signals.reserve(m_signals.size());
-  
-#if HAVE_SYS_EVENT_H
-  int queue_id = kqueue();
-  if (queue_id < 0)
-    Error("kqueue failed with error %d\n", queue_id);
   
-  std::vector<struct kevent> events;
-  events.reserve(m_read_fds.size() + m_signals.size());
-#else
-  sigset_t sigmask;
-  std::vector<struct pollfd> read_fds;
-  read_fds.reserve(m_read_fds.size());
-#endif
+  Error error;
+  auto impl = RunImpl::Create(*this, error);
+  if (!impl)
+    return error;
 
   // run until termination or until we run out of things to listen to
   while (!m_terminate_request && (!m_read_fds.empty() || !m_signals.empty())) {
-    // To avoid problems with callbacks changing the things we're supposed to
-    // listen to, we
-    // will store the *real* list of events separately.
-    signals.clear();
 
-#if HAVE_SYS_EVENT_H
-    events.resize(m_read_fds.size() + m_signals.size());
-    int i = 0;
-    for (auto &fd: m_read_fds) {
-      EV_SET(&events[i++], fd.first, EVFILT_READ, EV_ADD, 0, 0, 0);
-    }
-
-    for (const auto &sig : m_signals) {
-      signals.push_back(sig.first);
-      EV_SET(&events[i++], sig.first, EVFILT_SIGNAL, EV_ADD, 0, 0, 0);
-    }
-
-    struct kevent event_list[4];
-    int num_events =
-        kevent(queue_id, events.data(), events.size(), event_list, 4, NULL);
-
-    if (num_events < 0)
-      return Error("kevent() failed with error %d\n", num_events);
-
-#else
-    read_fds.clear();
-
-#if !SIGNAL_POLLING_UNSUPPORTED
-    if (int ret = pthread_sigmask(SIG_SETMASK, nullptr, &sigmask))
-      return Error("pthread_sigmask failed with error %d\n", ret);
-
-    for (const auto &sig : m_signals) {
-      signals.push_back(sig.first);
-      sigdelset(&sigmask, sig.first);
-    }
-#endif
-
-    for (const auto &fd : 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);
-#endif
-
-    for (int sig : signals) {
-      if (g_signal_flags[sig] == 0)
-        continue; // No signal
-      g_signal_flags[sig] = 0;
+    error = impl->Poll();
+    if (error.Fail())
+      return error;
 
+    impl->ForEachSignal([&](int sig) {
       auto it = m_signals.find(sig);
-      if (it == m_signals.end())
-        continue; // Signal must have gotten unregistered in the meantime
-
+      if (it != m_signals.end())
         it->second.callback(*this); // Do the work
-
-      if (m_terminate_request)
-        return Error();
-    }
+    });
 
-#if HAVE_SYS_EVENT_H
-    for (int i = 0; i < num_events; ++i) {
-      auto it = m_read_fds.find(event_list[i].ident);
-#else
-    for (auto fd : read_fds) {
-      if ((fd.revents & POLLIN) == 0)
-        continue;
-
-      auto it = m_read_fds.find(fd.fd);
-#endif
-      if (it == m_read_fds.end())
-        continue; // File descriptor must have gotten unregistered in the
-                  // meantime
+    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();
 }