[WIP][lldb] Add "event" capability to the MainLoop class

[mgorny]

To be honest, I'm not sure if "event" is the best name for it. I get it's a WSA name but it's a bit non-obvious to outsiders (I mean, technically everything is an event). Not that I have a better name in mind.

lldb/include/lldb/Host/MainLoop.h

66 ↗

(On Diff #449941)

Could you include a short comment explaining how it's used? I suppose something like "will be executed upon being explicitly signaled via NotifyEvent() from any thread".

lldb/source/Host/common/MainLoop.cpp

121 ↗

(On Diff #449941)

I think the reliance on < vs <= here could be confusing. Perhaps add a note that we popped the "event event" above, or maybe add a last_event const-var prior to popping it.

133 ↗

(On Diff #449941)

Let's make sure something didn't corrupt the state.

[mgorny]

BTW do you need me to implement the POSIX counterpart to this?

[labath]

In D131160#3701839, @mgorny wrote:

To be honest, I'm not sure if "event" is the best name for it. I get it's a WSA name but it's a bit non-obvious to outsiders (I mean, technically everything is an event). Not that I have a better name in mind.

Yeah, I know... I checked the thesaurus for "event", and I didn't find anything reasonable. And "signal" is obviously a bad idea..

Another possibility (one that just occurred to me right now) would be to use a completely different paradigm to achieve the effect. For example, we could extend the PendingCallback mechanism to enable adding callbacks from different threads (and have them wake the loop if necessary). That would be a more generic mechanism, and it would avoid the need to introduce a new concept (and having to name it).

In D131160#3701907, @mgorny wrote:

BTW do you need me to implement the POSIX counterpart to this?

That would be appreciated, but I think we first need to figure out the API, and how this code is going to be structured (I'm going to try handling the second part now)...

[mgorny]

Rebase on top of D131159 and partially implement the Linux counterpart.

Note that I have only covered sys/event.h for now and that I haven't tested the updated Windows code.

[mgorny]

lldb/source/Host/posix/MainLoopPosix.cpp

74–82

I'm wondering whether I should be repeating this all over the place or perhaps just adding m_event_pipe to m_read_fds.

105–109

Also I'm wondering about adding a ProcessEvents() wrapper that would take care of this.

[mgorny]

Ok, I've just realized that the pipe writing solution isn't going to work when the notification happens from the same thread (as the unittest does right now). @labath, do you happen to have an idea how to solve that reasonably cleanly?

[mgorny]

In D131160#3726642, @mgorny wrote:

Ok, I've just realized that the pipe writing solution isn't going to work when the notification happens from the same thread (as the unittest does right now). @labath, do you happen to have an idea how to solve that reasonably cleanly?

Ah, nevermind, for some reason I thought pipe writes are going to block until read happens. Perhaps we don't need to worry about that.

[mgorny]

Add m_event_pipe handler to m_read_fds to avoid reinventing the wheel three times.

[mgorny]

Add a test that notifying actually works on top of pending read.

[labath]

Thanks for taking care of the posix part.

I'm sort of warming up to the idea of reusing pending callback machinery for this. Besides avoiding the awkward naming problem, it also avoid another somewhat strange situation where our EventHandle offer additional actions on the event (which is different from all our other handle types, which only support deletion). All it would take is to add some locking around the PendingCallback manipulations, and write to the pipe (or trigger the event) whenever a new callback is registered, so that the Run function can invoke it.

It's a slightly different programming paradigm, and I suppose it also depends on which mechanism would be more natural for our InterruptRead use case (have you tried implementing it using this?), but I think the two mechanisms should be equivalent (as in, one could implement one using the other -- fairly easily).

What do you think about that?

In D131160#3726698, @mgorny wrote:

In D131160#3726642, @mgorny wrote:

Ok, I've just realized that the pipe writing solution isn't going to work when the notification happens from the same thread (as the unittest does right now). @labath, do you happen to have an idea how to solve that reasonably cleanly?

Ah, nevermind, for some reason I thought pipe writes are going to block until read happens. Perhaps we don't need to worry about that.

A pipe has a fixed (on linux, it's configurable) capacity, and it should be sufficiently large that we don't have to worry about it. If we wanted to be super safe, we could make the writes non-blocking. It should be safe to ignore EAGAINs in this situation, as that means the pipe already contains some unprocessed bytes, and all of the pending notifications will be processed once the reading thread "catches up".

lldb/source/Host/posix/MainLoopPosix.cpp

229

I like this.

[mgorny]

In D131160#3732668, @labath wrote:

Thanks for taking care of the posix part.

I'm sort of warming up to the idea of reusing pending callback machinery for this. Besides avoiding the awkward naming problem, it also avoid another somewhat strange situation where our EventHandle offer additional actions on the event (which is different from all our other handle types, which only support deletion). All it would take is to add some locking around the PendingCallback manipulations, and write to the pipe (or trigger the event) whenever a new callback is registered, so that the Run function can invoke it.

To be honest, I don't have a strong opinion, so I leave the decision to you. I was also thinking about this but decided to avoid changing the paradigm while rebasing/finishing the existing code, especially that I can't test it on Windows. But I can do it if you like.

It's a slightly different programming paradigm, and I suppose it also depends on which mechanism would be more natural for our InterruptRead use case (have you tried implementing it using this?), but I think the two mechanisms should be equivalent (as in, one could implement one using the other -- fairly easily).

Actually, I've started experimenting on rewriting the read thread option in Communication using this (with the goal of using the thread both for reads and writes).

lldb/source/Host/posix/MainLoopPosix.cpp

229

Thanks. I definitely like it more than doing it three times too ;-).

[mgorny]

Ok, so I'm going to give it a shot today. Could you clarify one thing though?

All it would take is to add some locking around the PendingCallback manipulations

Are you thinking of locking directly in MainLoop classes, or in the caller? It'd feel a bit inconsistent if MainLoops provided locking around pending callback manipulations and not other methods. However, I suppose making everything thread-safe wouldn't hurt either.

[labath]

In D131160#3735024, @mgorny wrote:

Ok, so I'm going to give it a shot today. Could you clarify one thing though?

All it would take is to add some locking around the PendingCallback manipulations

Are you thinking of locking directly in MainLoop classes, or in the caller? It'd feel a bit inconsistent if MainLoops provided locking around pending callback manipulations and not other methods. However, I suppose making everything thread-safe wouldn't hurt either.

I'm not sure this can be done from the outside, as the callback adittions need to synchronize with the accesses that happen inside the Run function. It is somewhat inconsistent, but for example python has special call_soon/call_soon_threadsafe versions for code which needs to schedule callbacks from other threads. We could do something like that. Or we could also just lock everything...

[mgorny]

Rebase. Replace "events" with triggering pending callbacks.

[mgorny]

@labath, what do you think of this?

Also I'm wondering about error handling. Should we introduce fallible constructor for this?

[labath]

What's the reasoning behind TriggerPendingCallbacks? I was assuming that the addition of a callback would cause it to run automatically...

[labath]

In D131160#3735372, @mgorny wrote:

Also I'm wondering about error handling. Should we introduce fallible constructor for this?

I'm not particularly worried about this. I might even make the Pipe constructor non-fallible, as the only way that can fail is if we exhaust our file descriptor limit -- and I very much doubt our ability to recover from that.

[mgorny]

In that case, is there something more I should do about this patch or are you going to take over from here?

[labath]

In D131160#3751959, @mgorny wrote:

In that case, is there something more I should do about this patch or are you going to take over from here?

It wasn't clear to me whether you intend to finish this patch (and I was waiting on your response to this). That said, if you don't have any strong opinions there, then I can try to finish this.

[mgorny]

In D131160#3738805, @labath wrote:

What's the reasoning behind TriggerPendingCallbacks? I was assuming that the addition of a callback would cause it to run automatically...

To be honest, I didn't think about it much. The original idea is that we can either have "non-important" callbacks that will be run at some point in the future, and "important" callbacks that should be run ASAP. However, I'm not really married to this idea, so I can make them triggered immediately if you prefer that.

In D131160#3761163, @labath wrote:

In D131160#3751959, @mgorny wrote:

In that case, is there something more I should do about this patch or are you going to take over from here?

It wasn't clear to me whether you intend to finish this patch (and I was waiting on your response to this). That said, if you don't have any strong opinions there, then I can try to finish this.

Ah, sorry, missed that comment. I can change that logic if you prefer. Otherwise, I think the next thing that needs to be here is testing/fixing the Windows code, and I'd be grateful if you could do that.

[labath]

Back to me.

[labath]

• Make TriggerPendingCallbacks protected
• fix the windows version

[labath]

In D131160#3763936, @mgorny wrote:

In D131160#3738805, @labath wrote:

What's the reasoning behind TriggerPendingCallbacks? I was assuming that the addition of a callback would cause it to run automatically...

To be honest, I didn't think about it much. The original idea is that we can either have "non-important" callbacks that will be run at some point in the future, and "important" callbacks that should be run ASAP. However, I'm not really married to this idea, so I can make them triggered immediately if you prefer that.

I don't think we need to make that distinction.

In D131160#3761163, @labath wrote:

In D131160#3751959, @mgorny wrote:

In that case, is there something more I should do about this patch or are you going to take over from here?

It wasn't clear to me whether you intend to finish this patch (and I was waiting on your response to this). That said, if you don't have any strong opinions there, then I can try to finish this.

Ah, sorry, missed that comment. I can change that logic if you prefer. Otherwise, I think the next thing that needs to be here is testing/fixing the Windows code, and I'd be grateful if you could do that.

I've fixed the windows version, and I've removed TriggerPendingCallbacks (from the public interface).

I think this should be good to go now.

[mgorny]

LGTM. I've just rebased and tested my patches on top of this, and they seem to work.

[mgorny]

I think this may have broken tests on Darwin.

[labath]

Are you sure about that? https://green.lab.llvm.org/green/view/LLDB/job/lldb-cmake/46639 is just running the commit now. I'd actually suspect the breakage is caused by D133181.

[mgorny]

In D131160#3771693, @labath wrote:

Are you sure about that? https://green.lab.llvm.org/green/view/LLDB/job/lldb-cmake/46639 is just running the commit now. I'd actually suspect the breakage is caused by D133181.

Ah, sorry, I was just guessing since it looked relevant.

lldb/include/lldb/Host/MainLoopBase.h

 //===-- MainLoopBase.h ------------------------------------------*- C++ -*-===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #ifndef LLDB_HOST_MAINLOOPBASE_H
 #define LLDB_HOST_MAINLOOPBASE_H
 
 #include "lldb/Utility/IOObject.h"
 #include "lldb/Utility/Status.h"
 #include "llvm/ADT/DenseMap.h"
 #include "llvm/Support/ErrorHandling.h"
 #include <functional>
+#include <mutex>
 
 namespace lldb_private {
 
 // The purpose of this class is to enable multiplexed processing of data from
 // different sources without resorting to multi-threading. Clients can register
 // IOObjects, which will be monitored for readability, and when they become
 // ready, the specified callback will be invoked. Monitoring for writability is
 // not supported, but can be easily added if needed.
public:
 
   virtual ReadHandleUP RegisterReadObject(const lldb::IOObjectSP &object_sp,
                                           const Callback &callback,
                                           Status &error) = 0;
 
   // 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.
   virtual Status Run() { llvm_unreachable("Not implemented"); }
 
   // This should only be performed from a callback. Do not attempt to terminate
   // the processing from another thread.
   virtual void RequestTermination() { m_terminate_request = true; }
 
 protected:
   ReadHandleUP CreateReadHandle(const lldb::IOObjectSP &object_sp) {
     return ReadHandleUP(new ReadHandle(*this, object_sp->GetWaitableHandle()));
   }
 
   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;
 
 private:
   class ReadHandle {
   public:
     ~ReadHandle() { m_mainloop.UnregisterReadObject(m_handle); }
 

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

 //===-- MainLoopPosix.h -----------------------------------------*- C++ -*-===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #ifndef LLDB_HOST_POSIX_MAINLOOPPOSIX_H
 #define LLDB_HOST_POSIX_MAINLOOPPOSIX_H
 
 #include "lldb/Host/Config.h"
 #include "lldb/Host/MainLoopBase.h"
+#include "lldb/Host/Pipe.h"
 #include "llvm/ADT/DenseMap.h"
 #include <csignal>
 #include <list>
 #include <vector>
 
 namespace lldb_private {
 
 // Implementation of the MainLoopBase class. It can monitor file descriptors for
SignalHandleUP RegisterSignal(int signo, const Callback &callback,
                                 Status &error);
 
   Status Run() override;
 
 protected:
   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);
 
   class SignalHandle {
   public:
     ~SignalHandle() { m_mainloop.UnregisterSignal(m_signo, m_callback_it); }
 
struct SignalInfo {
     std::list<Callback> callbacks;
     struct sigaction old_action;
     bool was_blocked : 1;
   };
   class RunImpl;
 
   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
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_HOST_POSIX_MAINLOOPPOSIX_H

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

 
 namespace lldb_private {
 
 // Windows-specific implementation of the MainLoopBase class. It can monitor
 // socket descriptors for readability using WSAEventSelect. Non-socket file
 // 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,
                                   Status &error) override;
 
   Status Run() override;
 
 protected:
   void UnregisterReadObject(IOObject::WaitableHandle handle) override;
 
+  void TriggerPendingCallbacks() override;
+
 private:
   void ProcessReadObject(IOObject::WaitableHandle handle);
   llvm::Expected<size_t> Poll();
 
   struct FdInfo {
     void *event;
     Callback callback;
   };
   llvm::DenseMap<IOObject::WaitableHandle, FdInfo> m_read_fds;
+  void *m_trigger_event;
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_HOST_WINDOWS_MAINLOOPWINDOWS_H

lldb/source/Host/common/MainLoopBase.cpp

 //===-- MainLoopBase.cpp --------------------------------------------------===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #include "lldb/Host/MainLoopBase.h"
 
 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

 //===-- MainLoopPosix.cpp -------------------------------------------------===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #include "lldb/Host/posix/MainLoopPosix.h"
 #include "lldb/Host/Config.h"
 #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>
 #include <csignal>
 #include <ctime>
 #include <vector>
 
 // Multiplexing is implemented using kqueue on systems that support it (BSD

 #endif
 
   sigset_t get_sigmask();
 #endif
 };
 
 #if HAVE_SYS_EVENT_H
 MainLoopPosix::RunImpl::RunImpl(MainLoopPosix &loop) : loop(loop) {
   in_events.reserve(loop.m_read_fds.size());
 }
 
 Status MainLoopPosix::RunImpl::Poll() {
   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);
 
   num_events = kevent(loop.m_kqueue, in_events.data(), in_events.size(),
                       out_events, llvm::array_lengthof(out_events), nullptr);
 
   if (num_events < 0) {
     if (errno == EINTR) {
       // in case of EINTR, let the main loop run one iteration
       // we need to zero num_events to avoid assertions failing
       num_events = 0;
     } else
       return Status(errno, eErrorTypePOSIX);
   }
   return Status();
 }
 
 void MainLoopPosix::RunImpl::ProcessEvents() {
   assert(num_events >= 0);
   for (int i = 0; i < num_events; ++i) {
     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");
     }
   }
 }
 #else
 MainLoopPosix::RunImpl::RunImpl(MainLoopPosix &loop) : loop(loop) {
 #ifndef __ANDROID__
   read_fds.reserve(loop.m_read_fds.size());
if (loop.m_terminate_request)
       return;
     g_signal_flags[signal] = 0;
     loop.ProcessSignal(signal);
   }
 }
 #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);
 #endif
 }
 
 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);
 }
 
 MainLoopPosix::ReadHandleUP
 MainLoopPosix::RegisterReadObject(const IOObjectSP &object_sp,
                                  const Callback &callback, Status &error) {
   if (!object_sp || !object_sp->IsValid()) {

 
 Status MainLoopPosix::Run() {
   m_terminate_request = false;
 
   Status 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())) {
-
+  // (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;
 
     impl.ProcessEvents();
 
     ProcessPendingCallbacks();
   }
if (it != m_signals.end()) {
     // The callback may actually register/unregister signal handlers,
     // so we need to create a copy first.
     llvm::SmallVector<Callback, 4> callbacks_to_run{
         it->second.callbacks.begin(), it->second.callbacks.end()};
     for (auto &x : callbacks_to_run)
       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

 #include <csignal>
 #include <ctime>
 #include <vector>
 #include <winsock2.h>
 
 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(),
                                  "WSAWaitForMultipleEvents failed");
 }
 
 MainLoopWindows::ReadHandleUP
 MainLoopWindows::RegisterReadObject(const IOObjectSP &object_sp,
Status MainLoopWindows::Run() {
 
   // run until termination or until we run out of things to listen to
   while (!m_terminate_request && !m_read_fds.empty()) {
 
     llvm::Expected<size_t> signaled_event = Poll();
     if (!signaled_event)
       return Status(signaled_event.takeError());
 
+    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

auto handle = loop.RegisterReadObject(
       error);
   ASSERT_TRUE(error.Success());
   ASSERT_TRUE(handle);
   ASSERT_TRUE(loop.Run().Success());
   // 2 iterations of read callback + 1 call of pending callback.
   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) {
 
   PseudoTerminal term;
   ASSERT_THAT_ERROR(term.OpenFirstAvailablePrimary(O_RDWR), llvm::Succeeded());
   ASSERT_THAT_ERROR(term.OpenSecondary(O_RDWR | O_NOCTTY), llvm::Succeeded());
   auto conn = std::make_unique<ConnectionFileDescriptor>(
       term.ReleasePrimaryFileDescriptor(), true);