[lldb] Enable the insertion of "pending callbacks" to MainLoops from …

…other threads

This will be used as a replacement for selecting over a pipe fd, which
does not work on windows. The posix implementation still uses a pipe
under the hood, while the windows version uses windows event handles.

The idea is that, instead of writing to a pipe, one just inserts a
callback, which does whatever you wanted to do after the bytes come out
the read end of the pipe.

Differential Revision: https://reviews.llvm.org/D131160

lldb/include/lldb/Host/MainLoopBase.h

@@ -14,6 +14,7 @@
 #include "llvm/ADT/DenseMap.h"
 #include "llvm/Support/ErrorHandling.h"
 #include <functional>
+#include <mutex>
 
 namespace lldb_private {
 
@@ -51,9 +52,7 @@ class MainLoopBase {
   // Add a pending callback that will be executed once after all the pending
   // events are processed. The callback will be executed even if termination
   // was requested.
-  virtual void AddPendingCallback(const Callback &callback) {
-    m_pending_callbacks.push_back(callback);
-  }
+  void AddPendingCallback(const Callback &callback);
 
   // Waits for registered events and invoke the proper callbacks. Returns when
   // all callbacks deregister themselves or when someone requests termination.
@@ -70,8 +69,13 @@ class MainLoopBase {
 
   virtual void UnregisterReadObject(IOObject::WaitableHandle handle) = 0;
 
+  // Interrupt the loop that is currently waiting for events and execute
+  // the current pending callbacks immediately.
+  virtual void TriggerPendingCallbacks() = 0;
+
   void ProcessPendingCallbacks();
 
+  std::mutex m_callback_mutex;
   std::vector<Callback> m_pending_callbacks;
   bool m_terminate_request : 1;
 

lldb/include/lldb/Host/posix/MainLoopPosix.h

@@ -11,6 +11,7 @@
 
 #include "lldb/Host/Config.h"
 #include "lldb/Host/MainLoopBase.h"
+#include "lldb/Host/Pipe.h"
 #include "llvm/ADT/DenseMap.h"
 #include <csignal>
 #include <list>
@@ -52,6 +53,8 @@ class MainLoopPosix : public MainLoopBase {
   void UnregisterReadObject(IOObject::WaitableHandle handle) override;
   void UnregisterSignal(int signo, std::list<Callback>::iterator callback_it);
 
+  void TriggerPendingCallbacks() override;
+
 private:
   void ProcessReadObject(IOObject::WaitableHandle handle);
   void ProcessSignal(int signo);
@@ -83,6 +86,7 @@ class MainLoopPosix : public MainLoopBase {
 
   llvm::DenseMap<IOObject::WaitableHandle, Callback> m_read_fds;
   llvm::DenseMap<int, SignalInfo> m_signals;
+  Pipe m_trigger_pipe;
 #if HAVE_SYS_EVENT_H
   int m_kqueue;
 #endif

lldb/include/lldb/Host/windows/MainLoopWindows.h

@@ -22,7 +22,8 @@ namespace lldb_private {
 // descriptors are not supported.
 class MainLoopWindows : public MainLoopBase {
 public:
-  ~MainLoopWindows() override { assert(m_read_fds.empty()); }
+  MainLoopWindows();
+  ~MainLoopWindows() override;
 
   ReadHandleUP RegisterReadObject(const lldb::IOObjectSP &object_sp,
                                   const Callback &callback,
@@ -33,6 +34,8 @@ class MainLoopWindows : public MainLoopBase {
 protected:
   void UnregisterReadObject(IOObject::WaitableHandle handle) override;
 
+  void TriggerPendingCallbacks() override;
+
 private:
   void ProcessReadObject(IOObject::WaitableHandle handle);
   llvm::Expected<size_t> Poll();
@@ -42,6 +45,7 @@ class MainLoopWindows : public MainLoopBase {
     Callback callback;
   };
   llvm::DenseMap<IOObject::WaitableHandle, FdInfo> m_read_fds;
+  void *m_trigger_event;
 };
 
 } // namespace lldb_private

lldb/source/Host/common/MainLoopBase.cpp

@@ -11,8 +11,23 @@
 using namespace lldb;
 using namespace lldb_private;
 
+void MainLoopBase::AddPendingCallback(const Callback &callback) {
+  {
+    std::lock_guard<std::mutex> lock{m_callback_mutex};
+    m_pending_callbacks.push_back(callback);
+  }
+  TriggerPendingCallbacks();
+}
+
 void MainLoopBase::ProcessPendingCallbacks() {
-  for (const Callback &callback : m_pending_callbacks)
+  // Move the callbacks to a local vector to avoid keeping m_pending_callbacks
+  // locked throughout the calls.
+  std::vector<Callback> pending_callbacks;
+  {
+    std::lock_guard<std::mutex> lock{m_callback_mutex};
+    pending_callbacks = std::move(m_pending_callbacks);
+  }
+
+  for (const Callback &callback : pending_callbacks)
     callback(*this);
-  m_pending_callbacks.clear();
 }

lldb/source/Host/posix/MainLoopPosix.cpp

@@ -11,6 +11,7 @@
 #include "lldb/Host/PosixApi.h"
 #include "lldb/Utility/Status.h"
 #include "llvm/Config/llvm-config.h"
+#include "llvm/Support/Errno.h"
 #include <algorithm>
 #include <cassert>
 #include <cerrno>
@@ -222,6 +223,18 @@ void MainLoopPosix::RunImpl::ProcessEvents() {
 #endif
 
 MainLoopPosix::MainLoopPosix() {
+  Status error = m_trigger_pipe.CreateNew(/*child_process_inherit=*/false);
+  assert(error.Success());
+  const int trigger_pipe_fd = m_trigger_pipe.GetReadFileDescriptor();
+  m_read_fds.insert({trigger_pipe_fd, [trigger_pipe_fd](MainLoopBase &loop) {
+                       char c;
+                       ssize_t bytes_read = llvm::sys::RetryAfterSignal(
+                           -1, ::read, trigger_pipe_fd, &c, 1);
+                       assert(bytes_read == 1);
+                       UNUSED_IF_ASSERT_DISABLED(bytes_read);
+                       // NB: This implicitly causes another loop iteration
+                       // and therefore the execution of pending callbacks.
+                     }});
 #if HAVE_SYS_EVENT_H
   m_kqueue = kqueue();
   assert(m_kqueue >= 0);
@@ -232,6 +245,8 @@ MainLoopPosix::~MainLoopPosix() {
 #if HAVE_SYS_EVENT_H
   close(m_kqueue);
 #endif
+  m_read_fds.erase(m_trigger_pipe.GetReadFileDescriptor());
+  m_trigger_pipe.Close();
   assert(m_read_fds.size() == 0); 
   assert(m_signals.size() == 0);
 }
@@ -347,8 +362,9 @@ Status MainLoopPosix::Run() {
   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())) {
-
+  // (m_read_fds will always contain m_trigger_pipe fd, so check for > 1)
+  while (!m_terminate_request &&
+         (m_read_fds.size() > 1 || !m_signals.empty())) {
     error = impl.Poll();
     if (error.Fail())
       return error;
@@ -377,3 +393,12 @@ void MainLoopPosix::ProcessSignal(int signo) {
       x(*this); // Do the work
   }
 }
+
+void MainLoopPosix::TriggerPendingCallbacks() {
+  char c = '.';
+  size_t bytes_written;
+  Status error = m_trigger_pipe.Write(&c, 1, bytes_written);
+  assert(error.Success());
+  UNUSED_IF_ASSERT_DISABLED(error);
+  assert(bytes_written == 1);
+}

lldb/source/Host/windows/MainLoopWindows.cpp

@@ -21,28 +21,39 @@
 using namespace lldb;
 using namespace lldb_private;
 
+MainLoopWindows::MainLoopWindows() {
+  m_trigger_event = WSACreateEvent();
+  assert(m_trigger_event != WSA_INVALID_EVENT);
+}
+
+MainLoopWindows::~MainLoopWindows() {
+  assert(m_read_fds.empty());
+  BOOL result = WSACloseEvent(m_trigger_event);
+  assert(result == TRUE);
+}
+
 llvm::Expected<size_t> MainLoopWindows::Poll() {
-  std::vector<WSAEVENT> read_events;
-  read_events.reserve(m_read_fds.size());
+  std::vector<WSAEVENT> events;
+  events.reserve(m_read_fds.size() + 1);
   for (auto &[fd, info] : m_read_fds) {
     int result = WSAEventSelect(fd, info.event, FD_READ | FD_ACCEPT | FD_CLOSE);
     assert(result == 0);
     (void)result;
 
-    read_events.push_back(info.event);
+    events.push_back(info.event);
   }
+  events.push_back(m_trigger_event);
 
-  DWORD result = WSAWaitForMultipleEvents(
-      read_events.size(), read_events.data(), FALSE, WSA_INFINITE, FALSE);
+  DWORD result = WSAWaitForMultipleEvents(events.size(), events.data(), FALSE,
+                                          WSA_INFINITE, FALSE);
 
   for (auto &fd : m_read_fds) {
     int result = WSAEventSelect(fd.first, WSA_INVALID_EVENT, 0);
     assert(result == 0);
     (void)result;
   }
 
-  if (result >= WSA_WAIT_EVENT_0 &&
-      result < WSA_WAIT_EVENT_0 + read_events.size())
+  if (result >= WSA_WAIT_EVENT_0 && result <= WSA_WAIT_EVENT_0 + events.size())
     return result - WSA_WAIT_EVENT_0;
 
   return llvm::createStringError(llvm::inconvertibleErrorCode(),
@@ -109,10 +120,18 @@ Status MainLoopWindows::Run() {
     if (!signaled_event)
       return Status(signaled_event.takeError());
 
-    auto &KV = *std::next(m_read_fds.begin(), *signaled_event);
-
-    ProcessReadObject(KV.first);
+    if (*signaled_event < m_read_fds.size()) {
+      auto &KV = *std::next(m_read_fds.begin(), *signaled_event);
+      ProcessReadObject(KV.first);
+    } else {
+      assert(*signaled_event == m_read_fds.size());
+      WSAResetEvent(m_trigger_event);
+    }
     ProcessPendingCallbacks();
   }
   return Status();
 }
+
+void MainLoopWindows::TriggerPendingCallbacks() {
+  WSASetEvent(m_trigger_event);
+}

lldb/unittests/Host/MainLoopTest.cpp

@@ -144,6 +144,33 @@ TEST_F(MainLoopTest, PendingCallbackCalledOnlyOnce) {
   ASSERT_EQ(3u, callback_count);
 }
 
+TEST_F(MainLoopTest, PendingCallbackTrigger) {
+  MainLoop loop;
+  std::promise<void> add_callback2;
+  bool callback1_called = false;
+  loop.AddPendingCallback([&](MainLoopBase &loop) {
+    callback1_called = true;
+    add_callback2.set_value();
+  });
+  Status error;
+  auto socket_handle = loop.RegisterReadObject(
+      socketpair[1], [](MainLoopBase &) {}, error);
+  ASSERT_TRUE(socket_handle);
+  ASSERT_THAT_ERROR(error.ToError(), llvm::Succeeded());
+  bool callback2_called = false;
+  std::thread callback2_adder([&]() {
+    add_callback2.get_future().get();
+    loop.AddPendingCallback([&](MainLoopBase &loop) {
+      callback2_called = true;
+      loop.RequestTermination();
+    });
+  });
+  ASSERT_THAT_ERROR(loop.Run().ToError(), llvm::Succeeded());
+  callback2_adder.join();
+  ASSERT_TRUE(callback1_called);
+  ASSERT_TRUE(callback2_called);
+}
+
 #ifdef LLVM_ON_UNIX
 TEST_F(MainLoopTest, DetectsEOF) {